F5负载均衡 3 节点 Kafka 集群实验

分类：Kafka作者：十八里铺的胖乞丐

1.环境配置

主机 操作系统 IP地址 安装软件
kafka1 Centos7.5 192.168.106.160 JDK1.8、zookeeper、kafka
kafka2 Centos7.5 192.168.106.161 JDK1.8、zookeeper、kafka
kafka3 Centos7.5 192.168.106.162 JDK1.8、zookeeper、kafka
F5 BIGIP-12.1.6 192.168.202.22 BIGIP-12.1.6

 2.准备工作

1.关闭防火墙

 kafka1，kafka2，kafka3三台主机均执行以下操作

##关闭防火墙
systemctl stop firewalld
##禁止开机启动
systemctl disable firewalld

2.下载JDK

点击下载JDK:Java Downloads | Oracle

 我选择的是jdk-8u371-linux-x64.rpm

 

 3.关于zookeeper

 本次实验是用的kafka自带zookeeper

 4.下载kafka

 点击下载kafka：https://downloads.apache.org/kafka/3.0.2/kafka_2.12-3.0.2.tgz

 5.下载F5

 点击注册：MyF5 | Login

 然后根据流程注册申请License，然后下载.OVA镜像

 注意下载版本，我这里用的是BIGIP-12.1.6-0.0.9：

 3.环境安装

 1.安装JDK（在三台kafka主机操作）

上传JDK至主机kafka1根目录，然后执行代码安装，完成后验证是否安装成功

##安装JDK
yum install jdk-8u371-linux-x64.rpm

##验证安装成功
java -version

因为是yum安装，不需要自行配置环境变量，主机kafka2跟主机kafka3 相同操作。

 2.安装kafka（在三台kafka主机操作）

 1.创建kafka运行目录，然后将kafka上传至目录解压

##创建文件夹
mkdir /opt/kafka

##将kafka压缩包上传至此目录

##进入此目录并解压
cd /opt/fakfa
tar zxvf kafka_2.12-3.0.2.tgz

2.创建日志保存地址

##创建日志保存地址
mkdir -p /opt/zookeeper-log/{data,log}
mkdir -p /opt/kafka-log/{data,log}

3.修改zookeeper.properties配置文件

##进入配置文件所在目录
cd /opt/kafka/kafka_2.12-3.0.2/config/

##修改配置文件
vi zookeeper.properties

##改为以下内容：

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License,Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# 
#    http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing,software
# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# the directory where the snapshot is stored.
dataDir=/opt/zookeeper-log/data
dataLogDir=/opt/zookeeper-log/logo
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080
# 在后面添加这六行
tickTime=2000
initLimit=10
syncLimit=5
server.1=192.168.106.160:2888:3888
server.2=192.168.106.161:2888:3888
server.3=192.168.106.162:2888:3888

  以上步骤三台kafka主机操作配置均相同。

 4.修改server.properties配置文件

##进入配置文件所在目录
cd /opt/kafka/kafka_2.12-3.0.2/config/

##修改配置文件
vi server.properties

kafka1修改为：


###################### Server Basics ###################

# kafka1为0，kafka2为1，kafka3为2，以此类推
broker.id=0

####################### Socket Server Settings ######################

# 填写本机IP地址.
advertised.listeners=PLAINTEXT://192.168.106.160:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

############################# Log Basics #############################

# 填写刚刚创建的日志路径
log.dirs=/opt/fakfa-log

# 分区数写为3
num.partitions=3

num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
#副本数改为3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################


############################# Log Retention Policy #############################

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# 这里填写刚刚在zookeeper.properties添加的地址，以逗号隔开
zookeeper.connect=192.168.106.160:2181,192.168.106.161:2181,192.168.106.162:2181

zookeeper.connection.timeout.ms=18000

############################# Group Coordinator Settings #############################

group.initial.rebalance.delay.ms=0

 完成后命令行输入以下命令写入myid

echo "1" >/opt/zookeeper-log/data/myid

kafka2修改为：


###################### Server Basics ###################

# kafka1为0，kafka2为1，kafka3为2，以此类推
broker.id=1

####################### Socket Server Settings ######################

# 填写本机IP地址.
advertised.listeners=PLAINTEXT://192.168.106.161:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

############################# Log Basics #############################

# 填写刚刚创建的日志路径
log.dirs=/opt/fakfa-log

# 分区数写为3
num.partitions=3

num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
#副本数改为3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################


############################# Log Retention Policy #############################

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# 这里填写刚刚在zookeeper.properties添加的地址，以逗号隔开
zookeeper.connect=192.168.106.160:2181,192.168.106.162:2181

zookeeper.connection.timeout.ms=18000

############################# Group Coordinator Settings #############################

group.initial.rebalance.delay.ms=0

 完成后命令行输入以下命令写入myid

echo "2" >/opt/zookeeper-log/data/myid

kafka3修改为：


###################### Server Basics ###################

# kafka1为0，kafka2为1，kafka3为2，以此类推
broker.id=2

####################### Socket Server Settings ######################

# 填写本机IP地址.
advertised.listeners=PLAINTEXT://192.168.106.162:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

############################# Log Basics #############################

# 填写刚刚创建的日志路径
log.dirs=/opt/fakfa-log

# 分区数写为3
num.partitions=3

num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
#副本数改为3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################


############################# Log Retention Policy #############################

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# 这里填写刚刚在zookeeper.properties添加的地址，以逗号隔开
zookeeper.connect=192.168.106.160:2181,192.168.106.162:2181

zookeeper.connection.timeout.ms=18000

############################# Group Coordinator Settings #############################

group.initial.rebalance.delay.ms=0

 完成后命令行输入以下命令写入myid

echo "3" >/opt/zookeeper-log/data/myid

4.启动集群

1.启动zookeeper

三台主机均输入以下命令启动

 ##进入kafka目录
cd /opt/kafka/kafka_2.12-3.0.2/
##输入命令启动
bin/zookeeper-server-start.sh config/zookeeper.properties &

 2.启动kafka：

bin/kafka-server-start.sh config/server.properties &

3.输入jps查看是否启动成功

4.测试集群效果

将kafka1作为主机，kafka2跟3作为从机，在kafka1执行下列命令，创建一个test主题并查看

##创建主题
bin/kafka-topics.sh --bootstrap-server 192.168.106.160:9092 --create --topic test --partitions 1
##查看主题
bin/kafka-topics.sh --bootstrap-server 192.168.106.160:9092 --list

创建完成后，测试消息的订阅和发送功能：

# 在kafka1 上发布消息
bin/kafka-console-producer.sh --broker-list  192.168.106.160:9092 --topic test
 
 
#  在kafka2 上接收
bin/kafka-console-consumer.sh --bootstrap-server 192.168.106.161:9092 --topic test --from-beginning

 
 
#  在kafka2 上接收
bin/kafka-console-consumer.sh --bootstrap-server 192.168.106.162:9092 --topic test --from-beginning

实现效果： 

 

 

 

5.负载均衡实现 

     负载均衡的安装等流程在我以前的文章有所提及，再次篇不再赘述，如有不懂的朋友们可以评论或者私信小编，小编看到会帮助其解答疑惑。

 1.登陆后配置vlan

 2.创建完成后添加IP

 3.添加完IP后配置pools

 

 配置完是这样

 4.然后去配置virtual servers

 

 下拉修改，然后提交即可。

 5.保存后点击你新建virtual servers的名字进入配置，然后点击Resources，进入后将其添加至pools。至此负载均衡已经配置完成

6.测试负载均衡效果 

 我在virtual servers添加的虚拟ip是192.168.106.164，下面测试我将用此IP进行展示。

# 在kafka1 上发布消息
bin/kafka-console-producer.sh --broker-list  192.168.106.164:9092 --topic test
 
 
#  在kafka2 上接收
bin/kafka-console-consumer.sh --bootstrap-server 192.168.106.164:9092 --topic test --from-beginning

 
 
#  在kafka2 上接收
bin/kafka-console-consumer.sh --bootstrap-server 192.168.106.164:9092 --topic test --from-beginning

 可以看到现在将收发消息的ip都换成了192.168.106.164，并且现在也不分主从，三台主机随便都可以做收发端，下面是效果展示

 7.总结

 此次实验已经完美结束了，中间也有很多小问题，不过到最后都完美解决了，上述就是全部的实验流程，大家如果有什么不懂的可以留言或者私信小编，小编看到后会第一时间给大家解答。

原文地址：https://blog.csdn.net/weixin_45109326/article/details/130382100

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

Kafka的认证
文章浏览阅读4.1k次。kafka认证_kafka认证
Kafka之常用参数配置整理
文章浏览阅读4.8k次，点赞4次，收藏11次。kafka常用参数_kafka配置
Kafka生产者发送消息的流程
文章浏览阅读1.4k次，点赞25次，收藏10次。Kafka 生产者发送消息的流程涉及多个步骤，从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
第3、4章 Kafka 生产者 和 消费者 ——向 Kafka 写入数据 和读取数据
文章浏览阅读854次，点赞22次，收藏24次。点对点模型：适用于一对一的消息传递，具有高可靠性。发布/订阅模型：适用于广播消息给多个消费者，实现消息的广播。主题模型：适用于根据消息的主题进行灵活的过滤和匹配，处理复杂的消息路由需求。
kafka消息队列的初步探索
文章浏览阅读1.5k次，点赞2次，收藏3次。kafka 自动配置在KafkaAutoConfiguration
【可视化工具】Kafka-Offset Explorer使用
文章浏览阅读1.3w次，点赞6次，收藏33次。Offset Explorer（以前称为Kafka Tool）是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI，允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境：windows 10版本：2.2Kafka版本：Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ，Next。_offset explorer
【云原生进阶之PaaS中间件】第三章Kafka-4.3.2-broker网络模型
文章浏览阅读1.3k次，点赞12次，收藏19次。kafka broker 在启动的时候，会根据你配置的listeners 初始化它的网络组件，用来接收外界的请求，这个listeners你可能没配置过，它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议，监听哪个端口，如果我们没有特殊的要求的话，使用它默认的配置就可以了，顶多是修改下端口这块。
Kafka: 详解、使用教程和示例
文章浏览阅读1.3k次，点赞2次，收藏2次。Kafka 是一个强大的分布式流处理平台，用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例，你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者，从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构，Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
【Spring从成神到升仙系列 四】从源码分析 Spring 事务的来龙去脉
文章浏览阅读3.5k次，点赞42次，收藏56次。对于Java开发者而言，关于 Spring ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。本期 Spring 源码解析系列文章，将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误，将不再一行一行的带大家分析源码，我们将一些不重要的分当做黑盒处理，以便我们更快、更有效的阅读源码。废话不多说，发车！
Kafka篇——Kafka消费者端常见配置，涵盖自动手动提交offset、poll消息细节、健康状态检查、新消费组消费offset规则以及指定分区等技术点配置，全面无死角，一篇文章拿下！
文章浏览阅读1.1k次，点赞14次，收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下，会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量，那么此时消费者挂了，于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
分布式 - 消息队列Kafka：Kafka生产者架构和配置参数
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send（）方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000，即60秒。在默认情况下，生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的send()方法调用会等待内存空间被释放，如果在max.block.ms之后还没有可用空间，就抛出异常。_kafka producer 参数
Kafka快速入门(最新版3.6.0)
文章浏览阅读2.9k次，点赞3次，收藏10次。kafka解决通信问题_kafka3.6
nginx+rsyslog+kafka+clickhouse+grafana 实现nginx 网关监控
文章浏览阅读1.5k次，点赞9次，收藏11次。上面都配置完了之后可以先验证下，保证数据最终到ck，如果有问题，需要再每个节点调试，比如先调试nginx->rsyslog ，可以先不配置kafka 输出，配置为console或者文件输出都可以，具体这里就不写了。这里做了一个类型转换，因为nginx，request-time 单位是s，我想最终呈现在grafana 中是ms，所以这里做了转换，当然grafana中也可以做。kafka 相关部署这里不做赘述，只要创建一个topic 就可以。
Kafka中的enable-auto-commit和auto-commit-interval配置
文章浏览阅读1.4k次，点赞22次，收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
thingsboard规则链调用外部API——kafka
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
Kafka 简介
文章浏览阅读1.3k次，点赞18次，收藏22次。Kafka_简介
从kafka如何保证数据一致性看通常数据一致性设计
文章浏览阅读1.1k次，点赞16次，收藏14次。在数据库系统中有个概念叫事务，事务的作用是为了保证数据的一致性，意思是要么数据成功，要么数据失败，不存在数据操作了一半的情况，这就是数据的一致性。在很多系统或者组件中，很多场景都需要保证数据的一致性，有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性，比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
元数据管理-Atlas的介绍和使用(集成Hive、Solr、Kafka、Kerberos)
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
zookeeper + kafka
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息，并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
Kafka-Kraft 模式架构部署
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot