kafka集群环境搭建与管理-创新互联

这一节主要讲kafka 集群环境部署,

成都创新互联主要为客户提供服务项目涵盖了网页视觉设计、VI标志设计、全网整合营销推广、网站程序开发、HTML5响应式网站建设手机网站制作、微商城、网站托管及成都网站维护公司、WEB系统开发、域名注册、国内外服务器租用、视频、平面设计、SEO优化排名。设计、前端、后端三个建站步骤的完善服务体系。一人跟踪测试的建站服务标准。已经为阳光房行业客户提供了网站建设服务。

kafka 基础概念介绍与强化

1)Producer:消息生产者,就是向kafkabroker发消息的客户端;
2)Consumer:消息消费者,向kafkabroker取消息的客户端;
3)Topic:可以理解为一个队列;
4)Consumer Group(CG):这是kafka用来实现一个topic消息的广播(发给所有的consumer)
和单播(发给任意一个consumer)的手段。一个topic可以有多个CG。topic的消息会复制
(不是真的复制,是概念上的)到所有的CG,但每个partion只会把消息发给该CG中的一
个consumer。如果需要实现广播,只要每个consumer有一个独立的CG就可以了。要实现
单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需
要多次发送消息到不同的topic;

5)Broker:一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker
可以容纳多个topic;

6)Partition:为了实现扩展性,一个非常大的topic可以分布到多个broker(即服务器)上,
一个topic可以分为多个partition,每个partition是一个有序的队列。partition中的每条消息
都会被分配一个有序的id(offset)。kafka只保证按一个partition中的顺序将消息发给
consumer,不保证一个topic的整体(多个partition间)的顺序;

分区数越多,在一定程度上会提升消息处理的吞吐量,因为kafka是基于文件进行读写,因此也需要打开更多的文件句柄,也会增加一定的性能开销。
如果分区过多,那么日志分段也会很多,写的时候由于是批量写,其实就会变成随机写了,随机 I/O 这个时候对性能影响很大。所以一般来说 Kafka 不能有太多的 Partition。

7)replication-factor
 用来设置主题的副本数。每个主题可以有多个副本,副本位于集群中不同的broker上,也就是说副本的数量不能超过broker的数量,否则创建主题时会失败。

在创建Topic的时候,有两个参数是需要填写的,那就是partions和replication-factor。

8)Offset:kafka的存储文件都是按照offset.kafka来命名,用offset做名字的好处是方便查

找。例如你想找位于2049的位置,只要找到2048.kafka的文件即可。当然thefirstoffset就
是00000000000.kafka。

集群环境搭建

  • zookeeper 集群环境搭建
  1. 下载软件包

wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

  1. 解压缩

  2. 创建配置目录,修改文件

[root@es7-1-80 conf]# cat zoo.cfg
initLimit=10
syncLimit=5
dataDir=/data/zk/data
clientPort=2181
maxClientCnxns=0
server.1=172.16.0.80:2888:3888
server.2=172.16.0.91:2888:3888
server.3=172.16.0.92:2888:3888

echo 1 >> =/data/zk/data/myid

  1. 启动,制作标准启动服务

    cat /etc/systemd/system/zookeeper.service
    [Unit]
    Description=zookeeper.service
    After=network.target

[Service]
Type=forking
ExecStart=/usr/local/zookeeper/bin/zkServer.sh start
ExecStop=/usr/local/zookeeper/bin/zkServer.sh stop
ExecReload=/usr/local/zookeeper/bin/zkServer.sh restart
[Install]
WantedBy=multi-user.target

其他节点按照上述方式配置,并依次启动

集群状态查看

/usr/local/zookeeper/bin/zkServer.sh status

  • kafka 集群搭建
  1. 下载软件包

curl -LO https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.1.0/kafka_2.12-2.1.0.tgz

  1. 解压缩到所需位置在

  2. 创建配置目录,修改文件

broker.id=0 #
num.inetwork.threads=3 #broker 处理消息的大线程数,一般情况下不需要去修改
num.io.threads=8 # # broker处理磁盘IO 的线程数 ,数值应该大于你的硬盘数
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/kafka/kafka_logs

num.partitions=1 # 每个topic的分区个数,若是在topic创建时候没有指定的话 会被topic创建时的指定参数覆盖
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=172.16.0.91:2181,172.16.0.80:2181,172.16.0.92:2181
zookeeper.connection.timeout.ms=6000

  1. 启动,制作标准启动服务

[Unit]
Description=Apache Kafka server (broker)
After=network.target
After=syslog.target
After=zookeeper.target

[Service]
Type=forking
User=root
Group=root
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh

ExecReload=/bin/kill -HUP $MAINPID
KillMode=none

Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

5、将kafka 命令加入到环境变量中

export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin

其他节点按照上述方式部署,修改broker.id

配置Kafka外网IP地址
kafka0.10以下
修改server.properties 配置文件
advertised.host.name=xxxx
advertised.port=9092
kafka0.10以上
listeners=PLAINTEXT://0.0.0.0:9093 //绑定所有ip
advertised.listeners=PLAINTEXT://42.159.7.75:9093

  • kafka 常用操作

    创建主题

    ./kafka-topics.sh --zookeeper 172.16.0.80:2181 --create --topic test1 --replication-factor 1 --partitions 3

    在创建Topic的时候,有两个参数是需要填写的,那就是partions和replication-factor。

    删除主题

./kafka-topics.sh --zookeeper 172.16.0.80:2181 --delete --topic test1

查看主题列表

./kafka-topics.sh --zookeeper 172.16.0.80:2181:2181 --list

查看主题状态

kafka-topics.sh --zookeeper 172.16.0.80:2181 --describe --topic test2
Topic:test2 PartitionCount:3 ReplicationFactor:1 Configs:
Topic: test2 Partition: 0 Leader: 2 Replicas: 2 Isr: 2
Topic: test2 Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: test2 Partition: 2 Leader: 1 Replicas: 1 Isr: 1

修改主题分区

./kafka-topics.sh --zookeeper 172.16.0.80:2181 --create --topic test3 --replication-factor 1 --partitions 1

 kafka-topics.sh --zookeeper 172.16.0.80:2181  --alter  --topic test3 --partitions 3

  kafka-topics.sh --zookeeper 172.16.0.80:2181  --describe  --topic test3

Topic:test3 PartitionCount:3 ReplicationFactor:1 Configs:
Topic: test3 Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: test3 Partition: 1 Leader: 2 Replicas: 2 Isr: 2
Topic: test3 Partition: 2 Leader: 0 Replicas: 0 Isr: 0

生产数据

kafka-console-producer.sh --broker-list 172.16.0.80:9092 --topic test4

hi go
[2019-11-26 15:02:32,608] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {test4=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-11-26 15:02:32,713] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 2 : {test4=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-11-26 15:02:32,820] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 3 : {test4=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-11-26 15:02:32,927] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 4 : {test4=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClien

解决方法
port=9092
listeners=PLAINTEXT://172.16.0.92:9092 # 修改为对应的ip

重新生产数据

kafka-console-producer.sh --broker-list 172.16.0.80:9092 --topic test4

hello kafaka

kafka-console-consumer.sh --zookeeper 172.16.0.91:2181 --topic test4 --from-beginning

消费数据

kafka-console-consumer.sh --zookeeper 172.16.0.91:2181 --topic test4 --from-beginning

列出消费者主组

消费组 即 Consumer Group,应该算是 Kafka 比较有亮点的设计了。那么何谓 Consumer Group 呢?用一句话概括就是:Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制。

/kafka-consumer-groups.sh --bootstrap-server 172.16.0.80:9092 --list

获取新版本消费者群组testgroup的详细信息

/kafka-consumer-groups.sh --bootstrap-server 172.16.0.80:9092 --group testgroup --describe

另外有需要云服务器可以了解下创新互联cdcxhl.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。


分享名称:kafka集群环境搭建与管理-创新互联
地址分享:http://pwwzsj.com/article/egsdi.html