centos7搭建kafka集群-第二篇

好了，本篇开始部署kafka集群

Zookeeper集群搭建

注:Kafka集群是把状态保存在Zookeeper中的，首先要搭建Zookeeper集群(也可以用kafka自带的ZK，但不推荐)

1、软件环境

（3台服务器）

10.0.20.131

10.0.20.132

10.0.20.133

1、Linux服务器一台、三台、五台、（2*n+1），Zookeeper集群的工作是超过半数才能对外提供服务，3台中超过两台超过半数，允许1台挂掉 ，是否可以用偶数，其实没必要。

如果有四台那么挂掉一台还剩下三台服务器，如果在挂掉一个就不行了，这里记住是超过半数。

2.Zookeeper需要依赖java环境

2.配置及安装Zookeeper集群

1.安装jdk

yum install java-1.8.0 -y

2.下载Zookeeper

首先要注意在生产环境中目录结构要定义好，防止在项目过多的时候找不到所需的项目

#我的目录统一放在/opt下面
#首先创建Zookeeper项目目录
mkdir /opt/zookeeper #项目目录
mkdir /opt/zookeeper/zkdata #存放快照日志
mkdir /opt/zookeeper/zkdatalog#存放日志
#下载软件
cd /opt/zookeeper/

wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz

#解压软件
tar -zxvf zookeeper-3.4.10.tar.gz

3、修改配置文件

进入到解压好的目录里面的conf目录中，查看

#进入conf目录
/opt/zookeeper/zookeeper-3.4.10/conf
#查看
[[email protected]192.168.7.107]$ ll
-rw-rw-r-- 1 1001 1001  535 Mar 23  2017 configuration.xsl
-rw-rw-r-- 1 1001 1001 2161 Mar 23  2017 log4j.properties
-rw-rw-r-- 1 1001 1001 1053 Nov 10 10:12 zoo_sample.cfg

#zoo_sample.cfg  这个文件是官方给我们的zookeeper的样板文件，给他复制一份命名为zoo.cfg，zoo.cfg是官方指定的文件命名规则。

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkdata
dataLogDir=/opt/zookeeper/zkdatalog
clientPort=2181
server.1=10.0.20.131:2888:3888
server.2=10.0.20.132:2888:3888
server.3=10.0.20.133:2888:3888
#server.1 这个1是服务器的标识也可以是其他的数字， 表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里
#10.0.20.131为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888

配置文件解释：

#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文件

#server1
echo "1" > /opt/zookeeper/zkdata/myid
#server2
echo "2" > /opt/zookeeper/zkdata/myid
#server3
echo "3" > /opt/zookeeper/zkdata/myid

重要配置说明

1、myid文件和server.myid  在快照目录下存放的标识本台服务器的文件，他是整个zk集群用来发现彼此的一个重要标识。

2、zoo.cfg 文件是zookeeper配置文件 在conf目录里。

3、log4j.properties文件是zk的日志输出文件 在conf目录里用java写的程序基本上有个共同点日志都用log4j，来进行管理。

# Define some default values that can be overridden by system properties
zookeeper.root.logger=INFO, CONSOLE  #日志级别
zookeeper.console.threshold=INFO  #使用下面的console来打印日志
zookeeper.log.dir=.    #日志打印到那里，是咱们启动zookeeper的目录 （建议设置统一的日志目录路径）
zookeeper.log.file=zookeeper.log
zookeeper.log.threshold=DEBUG
zookeeper.tracelog.dir=.
zookeeper.tracelog.file=zookeeper_trace.log

#
# ZooKeeper Logging Configuration
#

# Format is "<default threshold> (, <appender>)+

# DEFAULT: console appender only
log4j.rootLogger=${zookeeper.root.logger}

# Example with rolling log file
#log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE

# Example with rolling log file and tracing
#log4j.rootLogger=TRACE, CONSOLE, ROLLINGFILE, TRACEFILE

#
# Log INFO level and above messages to the console
#
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Threshold=${zookeeper.console.threshold}
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n

# Add ROLLINGFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold}
log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file}

# Max log file size of 10MB
log4j.appender.ROLLINGFILE.MaxFileSize=10MB
# uncomment the next line to limit number of backup files
#log4j.appender.ROLLINGFILE.MaxBackupIndex=10

log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n

#
# Add TRACEFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.TRACEFILE=org.apache.log4j.FileAppender
log4j.appender.TRACEFILE.Threshold=TRACE
log4j.appender.TRACEFILE.File=${zookeeper.tracelog.dir}/${zookeeper.tracelog.file}

log4j.appender.TRACEFILE.layout=org.apache.log4j.PatternLayout
### Notice we are including log4j‘s NDC here (%x)
log4j.appender.TRACEFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L][%x] - %m%n

4、zkEnv.sh和zkServer.sh文件

zkServer.sh 主的管理程序文件

zkEnv.sh 是主要配置，zookeeper集群启动时配置环境变量的文件

5、还有一个需要注意

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个。

推荐使用第一种方法，对于运维人员来说，将日志清理工作独立出来，便于统一管理也更可控。

4、启动服务并查看

1、启动服务

#进入到Zookeeper的bin目录下
cd /opt/zookeeper/zookeeper-3.4.10/bin
#启动服务（3台都需要操作）
./zkServer.sh start

2、检查服务状态

#检查服务器状态
./zkServer.sh status

通过status就能看到状态：

./zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.10/bin/../conf/zoo.cfg  #配置文件
Mode: follower  #他是否为领导

zk集群一般只有一个leader，多个follower，主一般是相应客户端的读写请求，而从主同步数据，当主挂掉之后就会从follower里投票选举一个leader出来。

可以用“jps”查看zk的进程，这个是zk的整个工程的main

#执行命令jps
20348 Jps
4233 QuorumPeerMain

至此Zookeeper集群搭建完毕！下面开始kafka集群搭建

Kafka集群搭建

1、软件环境

1、linux一台或多台，大于等于2

2、已经搭建好的zookeeper集群

3、软件版本kafka_2.12-0.10.2.1.tgz

2、创建目录并下载安装软件&添加环境变量

#创建目录
mkdir /opt/kafka #创建项目目录
mkdir /opt/kafka/kafkalogs #创建kafka消息目录，主要存放kafka消息

#下载软件
cd /opt/kafka/wget http://mirror.bit.edu.cn/apache/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz

#解压软件
tar -zxvf kafka_2.12-0.10.2.1.tgz#添加环境变量在/etc/profile文件下增加

　export KAFKA_HOME=/opt/kafka/kafka_2.12-0.10.2.1

　export PATH=$KAFKA_HOME/bin:$PATH

使用source /etc/profile使配置生效

3、修改配置文件

进入到config目录

cd /opt/kafka/kafka_2.12-0.10.2.1/config/

主要关注：server.properties 这个文件即可，我们可以发现在目录下：

有很多文件，这里可以发现有Zookeeper文件，我们可以根据Kafka内带的zk集群来启动，但是建议使用独立的zk集群

-rw-r--r-- 1 root root  906 Apr 22  2017 connect-console-sink.properties
-rw-r--r-- 1 root root  909 Apr 22  2017 connect-console-source.properties
-rw-r--r-- 1 root root 2760 Apr 22  2017 connect-distributed.properties
-rw-r--r-- 1 root root  883 Apr 22  2017 connect-file-sink.properties
-rw-r--r-- 1 root root  881 Apr 22  2017 connect-file-source.properties
-rw-r--r-- 1 root root 1074 Apr 22  2017 connect-log4j.properties
-rw-r--r-- 1 root root 2061 Apr 22  2017 connect-standalone.properties
-rw-r--r-- 1 root root 1199 Apr 22  2017 consumer.properties
-rw-r--r-- 1 root root 4369 Apr 22  2017 log4j.properties
-rw-r--r-- 1 root root 1900 Apr 22  2017 producer.properties
-rw-r--r-- 1 root root 5766 Nov 10 12:46 server.properties
-rw-r--r-- 1 root root 1032 Apr 22  2017 tools-log4j.properties
-rw-r--r-- 1 root root 1023 Apr 22  2017 zookeeper.properties

修改server.properties配置文件：

broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
listeners=PLAINTEXT://10.0.20.131:9092 #kafka监听地址
num.network.threads=3 #这个是borker进行网络处理的线程数
num.io.threads=8  #这个是borker进行I/O处理的线程数
socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
log.dirs=/opt/data/kafka-logs #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
num.partitions=1 #默认的分区数，一个topic默认1个分区数
num.recovery.threads.per.data.dir=1
log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天
log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除
zookeeper.connect=10.0.20.131:2181,10.0.20.132:2181,10.0.20.133:2181#设置zookeeper的连接端口
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
zookeeper.connection.timeout.ms=6000 #连接zookeeper的超时时间

上面是参数的解释，实际的修改项为：

#broker.id=0  每台服务器的broker.id都不能相同

#hostname
listeners=PLAINTEXT://10.0.20.131:9092

#在log.retention.hours=168 下面新增下面三项
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880

#设置zookeeper的连接端口
zookeeper.connect=10.0.20.131:2181,10.0.20.131:2181,10.0.20.131:2181

4、启动Kafka集群并测试

1、启动服务

#从后台启动Kafka集群（3台都需要启动）

kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-0.10.2.1/config/server.properties

2、检查服务是否启动

#执行命令jps
20348 Jps
4233 QuorumPeerMain
18991 Kafka

3、创建Topic来验证是否创建成功

更多请看官方文档：http://kafka.apache.org/documentation.html

#创建Topic
kafka-topics.sh --create --zookeeper 10.0.20.131:2181 --replication-factor 2 --partitions 1 --topic shuaige
#解释
--replication-factor 2   #复制两份
--partitions 1 #创建1个分区
--topic #主题为shuaige

‘‘‘在一台服务器上创建一个发布者‘‘‘
#创建一个broker，发布者
kafka-console-producer.sh --broker-list 10.0.20.131:9092 --topic shuaige

‘‘‘在一台服务器上创建一个订阅者‘‘‘kafka-console-consumer.sh --zookeeper 10.0.20.132:2181 --topic shuaige --from-beginning

测试（在发布者那里发布消息看看订阅者那里是否能正常收到~）：

4、其他命令

大部分命令可以去官方文档查看

4.1、查看topic

kafka-topics.sh --list --zookeeper 10.0.20.131:2181
#就会显示我们创建的所有topic

4.2、查看topic状态

kafka-topics.sh --describe --zookeeper 10.0.20.121:2181 --topic shuaige
#下面是显示信息
Topic:ssports    PartitionCount:1    ReplicationFactor:2    Configs:
    Topic: shuaige    Partition: 0    Leader: 1    Replicas: 0,1    Isr: 1
#分区为为1  复制因子为2   他的  shuaige的分区为0
#Replicas: 0,1   复制的为0，1

OK ! kafka集群搭建完毕

5、其他说明标注

5.1、日志说明

默认kafka的日志是保存在/opt/kafka/kafka_2.12-0.10.2.1/logs目录下的，这里说几个需要注意的日志

server.log #kafka的运行日志
state-change.log  #kafka他是用zookeeper来保存状态，所以他可能会进行切换，切换的日志就保存在这里

controller.log #kafka选择一个节点作为“controller”,当发现有节点down掉的时候它负责在游泳分区的所有节点中选择新的leader,这使得Kafka可以批量的高效的管理所有分区节点的主从关系。如果controller down掉了，活着的节点中的一个会备切换为新的controller.

5.2、上面的大家你完成之后可以登录zk来查看zk的目录情况

#使用客户端进入zk
./zkCli.sh -server 10.0.20.131:2181  #默认是不用加’-server‘参数的因为我们修改了他的端口

#查看目录情况 执行“ls /”
[zk: 10.0.20.131:2181(CONNECTED) 0] ls /

#显示结果：[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch]
‘‘‘
上面的显示结果中：只有zookeeper是，zookeeper原生的，其他都是Kafka创建的
‘‘‘

#标注一个重要的
[zk: 10.0.20.131:2181(CONNECTED) 1] get /brokers/ids/0
{"jmx_port":-1,"timestamp":"1456125963355","endpoints":["PLAINTEXT://10.0.20.131:9092"],"host":"10.0.20.131","version":2,"port":9092}
cZxid = 0x1000001c1
ctime = Mon Nov 10 15:26:03 CST 2017
mZxid = 0x1000001c1
mtime = Mon Nov 10 15:26:03 CST 2017
pZxid = 0x1000001c1
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x152e40aead20016
dataLength = 139
numChildren = 0
[zk: 10.0.20.131:2181(CONNECTED) 2] 

#还有一个是查看partion
[zk: 10.0.20.131:2181(CONNECTED) 7] get /brokers/topics/shuaige/partitions/0
null
cZxid = 0x100000029
ctime = Mon Nov 10 15:26:03 CST 2017
mZxid = 0x100000029
mtime = Mon Nov 10 15:26:03 CST 2017
pZxid = 0x10000002a
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
[zk: 10.0.20.131:2181(CONNECTED) 8]
时间： 2024-09-27 08:08:26

centos7搭建kafka集群-第二篇的相关文章

centos7搭建kafka集群-第一篇

Kafka初识 1.Kafka使用背景 在我们大量使用分布式数据库.分布式计算集群的时候,是否会遇到这样的一些问题: 我们想分析下用户行为(pageviews),以便我们设计出更好的广告位 我想对用户的搜索关键词进行统计,分析出当前的流行趋势 有些数据,存储数据库浪费,直接存储硬盘效率又低 这些场景都有一个共同点: 数据是由上游模块产生,上游模块,使用上游模块的数据计算.统计.分析,这个时候就可以使用消息系统,尤其是分布式消息系统! 2.Kafka的定义 What is Kafka:它是一个分布

docker容器搭建kafka集群

Docker搭建kafka集群 ?  需求说明: 公司目前有三个环境,生产环境,测试和演示环境,再包括开发人员还有开发的环境,服务器上造成了一定的资源浪费,因为环境需要依赖zookeeper和kafka,redis这些服务,只要搭一个环境,所有东西都要重新搭一遍,所以搭建kafka集群,让大部分环境都连接一个集群,把单个的服务变成公共的,稳定并易于管理 ?  Kafka集群管理和状态保存是通过zookeeper来实现的,要先部署zk集群 ?  环境说明: centos系统安装docker,通过d

基于docker环境搭建kafka集群（三台真机之间）

环境基于docker,三台物理主机192.168.0.27.192.168.0.28.192.168.0.29,每台主机部署一个zookeeper节点,一个kafka节点,共三个zookeeper节点,三个kafka节点,容器之间的网络采用host模式 1.拉取镜像 2.启动容器 step1. 创建配置文件zoo.cfg并替换容器内的该文件,不同的镜像有可能zoo.cfg位置不同 # The number of milliseconds of each tick tickTime=2000 #

Centos7.4 kafka集群安装与kafka-eagle1.3.9的安装

Centos7.4 kafka集群安装与kafka-eagle1.3.9的安装 集群规划: hostname Zookeeper Kafka kafka-eagle kafka01 √ √ √ kafka02 √ √ kafka03 √ √ 准备工作: 网络配置 vim /etc/sysconfig/network-scripts/ifcfg-ens33 # kafka01 TYPE=Ethernet PROXY_METHOD=none BROWSER_ONLY=no BOOTPROTO=sta

docker容器中搭建kafka集群环境

Kafka集群管理.状态保存是通过zookeeper实现,所以先要搭建zookeeper集群 zookeeper集群搭建 一.软件环境: zookeeper集群需要超过半数的的node存活才能对外服务,所以服务器的数量应该是2*N+1,这里使用3台node进行搭建zookeeper集群. 1. 3台linux服务器都使用docker容器创建,ip地址分别为 NodeA:172.17.0.10 NodeB:172.17.0.11 NodeC:172.17.0.12 2. zookeeper的doc

Centos7搭建RabbitMQ集群

本片博客简单介绍rabbitmq单节点以及集群的配置安装,使用和rabbitmq的命令,和图形界面简单的介绍使用 话不多说直接进入正题,首先介绍下rabbitmq: rabbitmq的历史:RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue )的开源实现.AMQP 的出现其实也是应了广大人民群众的需求,虽然在同步消息通讯的世界里有很多公开标准(如 COBAR的 IIOP ,或者是 SOAP 等),但是在异步消息处理中却不是这样,只有大企业有一些商业实

搭建Kafka集群

1. 启动Zookeeper服务器  ./zookeeper-server-start.sh /opt/cx/kafka_2.11-0.9.0.1/config/zookeeper.properties 2. 修改broker-1,broker-2配置 broker.id=1 listeners=PLAINTEXT://:9093  # The port the socket server listens on port=9093 log.dirs=/opt/cx/kafka/kafka-log

centos7搭建zookeeper集群环境

搭建起一个zookeeper的集群环境,选择三台机器作为测试(zookeeper的集群环境节点数最好的是>=3且为奇数) 选用三台虚拟机机器(三台机器上需要配置好jdk环境): Node1:192.168.153.133 Node2:192.168.153.137 Node3:192.168.153.138 首先下载zookeeper的稳定版,地址https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/stable/apache-zookeep

五 搭建kafka集群

1 下载    wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.0.0/kafka_2.12-2.0.0.tgz 2 tar  -zxvf  kafka_2.12-2.0.0.tgz -C  kafka 3  进入到config下: 修改zookeeper.properties, 配置三台servers. 4: 创建myid文件,进入/usr/local/kafka/zookeeper,创建myid文件,将三个服务器上的myid文件