Zookeeper+Kafka集群部署

古城微笑少年丶 2022-01-12 23:05 402阅读 0赞

Kafka的基本概念：

主题：Topic特指Kafka处理的消息源（feeds of messages）的不同分类。

分区：Partition Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。

Message：消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。

Producers：消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。

Consumers：消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。

Broker：缓存代理，Kafka集群中的一台或多台服务器统称为broker，这里用的是AMQP协议。

AMQP协议：Advanced Message Queuing Protocol （高级消息队列协议）

是一个标准开放的应用层的消息中间件协议。AMQP定义了通过网络发送的字节流的数据格式。因此兼容性非常好，任何实现AMQP协议的程序都可以和与AMQP协议兼容的其他程序交互，可以很容易做到跨语言，跨平台。

Kafka相对传统技术的优势：

快速：单一的Kafka代理可以处理成千上万的客户端，每秒处理数兆字节的读写操作。

可伸缩：在一组机器上对数据进行分区和简化，以支持更大的数据

持久：消息是持久性的，并在集群中进行复制，以防止数据丢失。

设计：它提供了容错保证和持久性

Kafka的工作流程、架构图：

![k1.png][]

图1

![k2.png][]

图2

如图所示，Kafka的服务器数量一般为奇数（2n+1），这是因为Zookeeper集群的工作主机必须超过搭建集群的主机半数才能对外提供服务，所以，这里用奇数避免某台服务器故障导致集群不能正常工作。

因为Kafka集群是把状态保存在Zookeeper中的，并且Kafka的动态扩容是通过Zookeeper来实现的，所以需要优先搭建Zookeerper集群，建立分布式状态管理。捋清楚思路后，开始准备环境，搭建集群：

server1：192.168.11.141

server2：192.168.11.142

server3：192.168.11.143

zookeeper是基于Java环境开发的所以，需要先安装Java，然后这里使用的zookeeper安装包版本为zookeeper-3.4.14，Kafka的安装包版本为kafka\_2.11-2.2.0。

1.修改主机名，关闭selinux，防火墙。

hostnamectl set-hostname --static server1 && bash

setenforce 0

sed -i "s/SELINUX=enforcing/SELINUX=disabled/g" /etc/selinux/config

systemctl stop firewalld && systemctl disable firewalld

2.创建zookeeper和Kafka的存放目录：

cd /usr/local/

mkdir -pv zookeeper/\{zkdata,zkdatalog\}                 \#zkdata是存放快照日志，zkdatalog是存放事物日志

mkdir -pv kafka/kfdatalogs                 \#kfdatalogs是存放消息日志

3.下载软件包，并解压到各自的工作目录下（当前在/usr/local/下）：

yum -y install java-1.8.0\*                                         \#这里我的源用的是阿里的base源和epel源，可以直接装这个版本的java。

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

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka\_2.11-2.2.0.tgz

tar xzf zookeeper-3.4.14.tar.gz -C zookeeper

tar xzf kafka\_2.11-2.2.0.tgz -C kafka

4.生成并更改zookeeper的配置文件（三台服务器上都需要设置）：

\#可以将zoo\_sample理解为zookeeper自带的配置文件模板，复制出一份以.cfg结尾的配置文件。

cp -av /usr/local/zookeeper/zookeeper-3.4.14/conf/\{zoo\_sample,zoo.cfg\}

tickTime=2000              \#zookeeper服务器之间的心跳时间。

initLimit=10                 \#zookeeper的最大连接失败时间

syncLimit=5          \#zookeeper的同步通信时间

dataDir=/usr/local/zookeeper/zkdata                 \#zookeeper的存放快照日志的绝对路径

dataLogDir=/usr/local/zookeeper/zkdatalog                 \#zookeeper的存放事物日志的绝对路径

clientPort=2181              \#zookeeper与客户端的连接端口

server.1=192.168.11.139:2888:3888                  \#服务器及其编号，服务器IP地址，通信端口，选举端口

server.2=192.168.11.140:2888:3888                 \#服务器及其编号，服务器IP地址，通信端口，选举端口

server.3=192.168.11.141:2888:3888                \#服务器及其编号，服务器IP地址，通信端口，选举端口

\#以上端口都是zookeeper的默认端口，可随需求进行修改

5.创建myid文件：

在server1上

echo "1" > /usr/local/zookeeper/zkdata/myid

\#就是在不同的服务器上，将服务器编号发送到zkdata下的myid。

6.启动zookeeper集群

cd /usr/local/zookeeper/zookeeper-3.4.14/bin

./zkServer.sh start

./zkServer.sh status

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

到此，zookeeper集群搭建结束，接下来基于zookeeper搭建kafka集群：

cd /usr/local/kafka/kafka\_2.11-2.2.0/config/

1.修改server.properties文件：

broker.id=1              \#这里和zookeeper中的myid文件一样，采用的是唯一标识

prot=9092              \#Kafka集群间连接的端口，配置文件中没有，但默认为9092，可随需求进行修改，这里咱们加上

log.dirs=/usr/local/kafka/kfdatalogs                         \#存放Kafka消息日志的绝对路径

host.name=192.168.11.141                  \#顾名思义就是服务器的IP地址

log.retention.hours=168                  \#默认消息的最大持久化时间，168小时，7天

message.max.byte=5242880                      \#消息保存的最大值5M

default.replication.factor=2                         \#kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务

replica.fetch.max.bytes=5242880                 \#取消息的最大直接数

zookeeper.connect=192.168.11.141:2181,192.168.11.142:2181,192.168.11.143:2181\#集群的各个节点的IP地址及zookeeper的端口，在zookeeper集群设置的端口是多少这里的端口就是多少。

未修改的配置文件信息：

num.network.threads=3              \#这个是borker进行网络处理的线程数

num.io.threads=8                 \#这个是borker进行I/O处理的线程数

num.partitions=1                  \#默认的分区数，一个topic默认1个分区数

log.retention.hours=168                  \#默认消息的最大持久化时间，168小时，7天

message.max.byte=5242880               \#消息保存的最大值5M

default.replication.factor=2                          \#kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务

replica.fetch.max.bytes=5242880                          \#取消息的最大直接数

log.segment.bytes=1073741824                         \#这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件

log.retention.check.interval.ms=300000                  \#每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除

log.cleaner.enable=false                      \#是否启用log压缩，一般不用启用，启用的话可以提高性能

2.启动Kafka集群：

cd /usr/local/kafka/kafka\_2.11-2.2.0/bin

bash kafka-server-start.sh -daemon ../config/server.properties

jps                   \#查看一下

3.开始测试：

3.1创建topic

./kafka-topics.sh --create --zookeeper 192.168.21.241：2181 --replication-factor 2 --partitions 1 --topic tian

\#--replication-factor 2                  复制两份

\#--partitions 1                 创建一个分区

\#--topic tian                 主题为tian

3.2创建一个broker：

./kafka-console-producer.sh --broker-list 192.168.21.241:9092 --topic tian

3.3创建一个consumer：

./kafka-console-consumer.sh  --bootstrap-server 192.168.11.141:9092 --topic tian --from-beginning

3.4查看topic：

./kafka-topics.sh --list --zookeeper 192.168.11.141:12181

到此，Kafka集群的搭建完成。

转载于:https://blog.51cto.com/13860705/2394657

[k1.png]: https://s1.51cto.com/images/20190514/1557847735842298.png?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=
[k2.png]: https://s1.51cto.com/images/20190514/1557847800423955.png?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=