一、为啥用Kafka二、Zookeeper与Kafka以及SpringBoot的版本对应
2.1、JDK8支持的Kafka版本2.2、Kafka与SpringBoot的版本对应 三、Zookeeper集群搭建
3.1、软件环境3.2、配置&安装Zookeeper3.3启动并查看zookeeper 四、安装Kafka集群
一、为啥用Kafka 二、Zookeeper与Kafka以及SpringBoot的版本对应 2.1、JDK8支持的Kafka版本打开kafka的官网,找到对应版本的kafka,查看文档中各版本对jdk8(java8)的支持(下面地址是2.6的版本,注意选自己需要的版本)
文档地址:https://kafka.apache.org/26/documentation.html
本次集群我是用的版本是kafka_2.13-2.5.0,这个版本支持jdk1.8
使用的Zookeeper是kafka_2.13-2.5.0.
这里需要强调的是,先看自己电脑安装的java版本,然后去查能对应安装的kafka和Zookeeper的最高版本是什么,然后再对应Zookeeper能对应的kafka本本。
2.2、Kafka与SpringBoot的版本对应 三、Zookeeper集群搭建Kafka集群是把状态保存在Zookeeper中的,首先要搭建Zookeeper集群。
3.1、软件环境(3台服务器-我的测试)
192.168.1.221 server1
192.168.1.110 server2
192.168.1.31 server3
1、服务器一台、三台、五台、(2*n+1),Zookeeper集群的工作是超过半数才能对外提供服务,3台中超过两台超过半数,允许1台挂掉 ,是否可以用偶数,其实没必要。
如果有四台那么挂掉一台还剩下三台服务器,如果在挂掉一个就不行了,这里记住是超过半数。
2、Java jdk1.8 zookeeper是用java写的所以他的需要JAVA环境,java是运行在java虚拟机上的
3、Zookeeper的稳定版本Zookeeper 3.4.6版本
解压Zookeeper压缩包配置系统环境变量(系统变量中新增ZOOKEEPER_HOME变量,指向Zookeeper所在根目录
- 系统变量path中添加“%ZOOKEEPER_HOME%bin”修改配置文件
# The number of milliseconds of each tick tickTime=2000 # The number of ticks that the initial # synchronization phase can take initLimit=10 # The number of ticks that can pass between # sending a request and getting an acknowledgement syncLimit=5 # the directory where the snapshot is stored. # do not use /tmp for storage, /tmp here is just # example sakes. dataDir=E:\MOM\apache-zookeeper-3.7.0-bin\zkdata # 事物日志的存储路径,如果不配置这个那么事物日志会默认存储到dataDir制定的目录,这样会严重影响zk的性能,当zk吞吐量较大的时候,产生的事物日志、快照日志太多 dataLogDir=E:\MOM\apache-zookeeper-3.7.0-bin\zkdatalog # the port at which the clients will connect clientPort=2181 server.1=192.168.1.221:2888:3888 server.2=192.168.1.110:2888:3888 server.3=192.168.1.31:2888:3888 #server.1 这个1是服务器的标识也可以是其他的数字, 表示这个是第几号服务器,用来标识服务器,这个标识要写到快照目录下面myid文件里 #192.168.1.221为集群里的IP地址,第一个端口是master和slave之间的通信端口,默认是2888,第二个端口是leader选举的端口,集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888 #myid文件没有后缀,dataDir目录下直接创建名字myid,内容分别为server后的数字1、 2 、 3 # the maximum number of client connections. # increase this if you need to handle more clients #maxClientCnxns=60 # # Be sure to read the maintenance section of the # administrator guide before turning on autopurge. # # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance # # The number of snapshots to retain in dataDir #autopurge.snapRetainCount=3 # Purge task interval in hours # Set to "0" to disable auto purge feature #autopurge.purgeInterval=1 ## Metrics Providers # # https://prometheus.io Metrics Exporter #metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider #metricsProvider.httpPort=7000 #metricsProvider.exportJvmInfo=true
配置文件解释
#tickTime: 这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔,也就是每个 tickTime 时间就会发送一个心跳。 #initLimit: 这个配置项是用来配置 Zookeeper 接受客户端(这里所说的客户端不是用户连接 Zookeeper 服务器的客户端,而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器)初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间(也就是 tickTime)长度后 Zookeeper 服务器还没有收到客户端的返回信息,那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒 #syncLimit: 这个配置项标识 Leader 与Follower 之间发送消息,请求和应答时间长度,最长不能超过多少个 tickTime 的时间长度,总的时间长度就是5*2000=10秒 #dataDir: 快照日志的存储路径 #dataLogDir: 事物日志的存储路径,如果不配置这个那么事物日志会默认存储到dataDir制定的目录,这样会严重影响zk的性能,当zk吞吐量较大的时候,产生的事物日志、快照日志太多 #clientPort: 这个端口就是客户端连接 Zookeeper 服务器的端口,Zookeeper 会监听这个端口,接受客户端的访问请求。修改他的端口改大点
创建myid文件
在配置文件中指定的快照日志的存储路径(dataDir=E:\MOM\apache-zookeeper-3.7.0-bin\zkdata)文件夹中创建myid文件,创建文本文件,修改名字为myid,去掉后缀,server.1 服务器内容就写1然后保存,也就是说要把配置文件中server.后面指定的标识写入myid文件。注意每个服务器根据自己ip对应的标识写入这里还要强调一点,myid保存是最好不要修改文档编码格式,我改成utf-8后启动zookeeper后报错,改成默认就好了。
重要配置说明
myid文件和server.myid 在快照目录下存放的标识本台服务器的文件,他是整个zk集群用来发现彼此的一个重要标识。zoo.cfg 文件是zookeeper配置文件 在conf目录里。
此处引用一个转载说明:
ZooKeeper server will not remove old snapshots and log files when using the default configuration (see autopurge below), this is the responsibility of the operator
zookeeper不会主动的清除旧的快照和日志文件,这个是操作者的责任。
但是可以通过命令去定期的清理。
#!/bin/bash #snapshot file dir dataDir=/opt/zookeeper/zkdata/version-2 #tran log dir dataLogDir=/opt/zookeeper/zkdatalog/version-2 #Leave 66 files count=66 count=$[$count+1] ls -t $dataLogDir/log.* | tail -n +$count | xargs rm -f ls -t $dataDir/snapshot.* | tail -n +$count | xargs rm -f #以上这个脚本定义了删除对应两个目录中的文件,保留最新的66个文件,可以将他写到crontab中,设置为每天凌晨2点执行一次就可以了。 #zk log dir del the zookeeper log #logDir= #ls -t $logDir/zookeeper.log.* | tail -n +$count | xargs rm -f
其他方法:
第二种:使用ZK的工具类PurgeTxnLog,它的实现了一种简单的历史文件清理策略,可以在这里看一下他的使用方法 http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html
第三种:对于上面这个执行,ZK自己已经写好了脚本,在bin/zkCleanup.sh中,所以直接使用这个脚本也是可以执行清理工作的。
第四种:从3.4.0开始,zookeeper提供了自动清理snapshot和事务日志的功能,通过配置 autopurge.snapRetainCount 和 autopurge.purgeInterval 这两个参数能够实现定时清理了。这两个参数都是在zoo.cfg中配置的:
autopurge.purgeInterval 这个参数指定了清理频率,单位是小时,需要填写一个1或更大的整数,默认是0,表示不开启自己清理功能。
autopurge.snapRetainCount 这个参数和上面的参数搭配使用,这个参数指定了需要保留的文件数目。默认是保留3个。
推荐使用第一种方法,对于运维人员来说,将日志清理工作独立出来,便于统一管理也更可控。毕竟zk自带的一些工具并不怎么给力。
3.3启动并查看zookeeper在控制台输入:zkServer启动zookeeper
zk集群一般只有一个leader,多个follower,主一般是相应客户端的读写请求,而从主同步数据,当主挂掉之后就会从follower里投票选举一个leader出来。
可以用“jps”查看zk的进程,这个是zk的整个工程的main
四、安装Kafka集群主要是将kafka加压到目录,然后再配置文件中最好配置就可以启动了,
config/server.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. # see kafka.server.KafkaConfig for additional details and defaults ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. # 当前机器在集群中的唯一标识,和zookeeper的myid性质一样 broker.id=0 ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=PLAINTEXT://:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). advertised.listeners=PLAINTEXT://192.168.1.221:9092 # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL # The number of threads that the server uses for receiving requests from the network and sending responses to the network # 这个是borker进行网络处理的线程数 num.network.threads=3 # The number of threads that the server uses for processing requests, which may include disk I/O # 这个是borker进行I/O处理的线程数 num.io.threads=8 # 发送缓冲区buffer大小,数据不是一下子就发送的,先回存储到缓冲区了到达一定的大小后在发送,能提高性能 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer.bytes=102400 # kafka接收缓冲区大小,当数据到达一定大小后在序列化到磁盘 # The receive buffer (SO_RCVBUF) used by the socket server socket.receive.buffer.bytes=102400 # 这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数,这个值不能超过java的堆栈大小 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 ############################# Log Basics ############################# # 消息存放的目录,这个目录可以配置为“,”逗号分割的表达式,上面的num.io.threads要大于这个目录的个数这个目录 # 如果配置多个目录,新创建的topic他把消息持久化的地方是,当前以逗号分割的目录中,那个分区数最少就放那一个 # A comma separated list of directories under which to store log files log.dirs=D:\mom\kafka_2.13-2.5.0\kafka-data # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. # 默认的分区数,一个topic默认1个分区数 num.partitions=1 # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. num.recovery.threads.per.data.dir=1 ############################# Internal Topic Settings ############################# # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 ############################# Log Flush Policy ############################# # Messages are immediately written to the filesystem but by default we only fsync() to sync # the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs here: # 1. Durability: Unflushed data may be lost if you are not using replication. # 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. # 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or # every N messages (or both). This can be done globally and overridden on a per-topic basis. # The number of messages to accept before forcing a flush of data to disk #log.flush.interval.messages=10000 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 ############################# Log Retention Policy ############################# # The following configurations control the disposal of log segments. The policy can # be set to delete segments after a period of time, or after a given size has accumulated. # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens # from the end of the log. # 默认消息的最大持久化时间,168小时,7天 # The minimum age of a log file to be eligible for deletion due to age log.retention.hours=168 # 消息保存的最大值5M message.max.byte=5242880 # kafka保存消息的副本数,如果一个副本失效了,另一个还可以继续提供服务 default.replication.factor=2 # 取消息的最大直接数 replica.fetch.max.bytes=5242880 # A size-based retention policy for logs. Segments are pruned from the log unless the remaining # segments drop below log.retention.bytes. Functions independently of log.retention.hours. #log.retention.bytes=1073741824 # 这个参数是:因为kafka的消息是以追加的形式落地到文件,当超过这个值的时候,kafka会新起一个文件 # The maximum size of a log segment file. When this size is reached a new log segment will be created. log.segment.bytes=1073741824 # 每隔300000毫秒去检查上面配置的log失效时间(log.retention.hours=168 ),到目录查看是否有过期的消息如果有,删除 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 ############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=192.168.1.221:2181,192.168.1.110:2181,192.168.1.31:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=18000 ############################# Group Coordinator Settings ############################# # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. # The default value for this is 3 seconds. # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. group.initial.rebalance.delay.ms=0
这里说明一下,主要修改四个指标:
broker.id=0 # 当前机器在集群中的唯一标识,和zookeeper的myid性质一样三台机器要设置不同的broker.id listeners=PLAINTEXT://:9092 # 每台服务器要根据自己的实际ip进行修改 advertised.listeners=PLAINTEXT://192.168.1.221:9092 # 消息存放的目录,这个目录可以配置为“,”逗号分割的表达式,上面的num.io.threads要大于这个目录的个数这个目录 # 如果配置多个目录,新创建的topic他把消息持久化的地方是,当前以逗号分割的目录中,那个分区数最少就放那一个 log.dirs=D:\mom\kafka_2.13-2.5.0\kafka-data # 需要将zookeeper服务器地址及端口全部配置上 zookeeper.connect=192.168.1.221:2181,192.168.1.110:2181,192.168.1.31:2181
此时启动kafka就可以了,注意需要先启动zookeeper



