kafka高可用集群环境搭建

kafka是分布式发布订阅消息系统

Kafka作为一个集群，运行在一台或者多台服务器上.

Kafka 通过 topic 对存储的流数据进行分类。

每条记录中包含一个key，一个value和一个timestamp（时间戳）。

中文官方网站：http://kafka.apachecn.org/quickstart.html

优点：

a、高吞吐量、低延迟：kafka每秒可以处理几十万条消息，延迟最低只有几毫秒，每个topic可以分多个partition、consumer group 对partition进行consumer操作

b、可扩展性：集群支持可扩展性

c、持久性：消息被持久化到本地磁盘，支持数据备份

d、高并发：支持数千个客户端同时进行读写操作

kafka存储

kafka是以topic来进行消息管理，每个topic包含多个partition，每个partition对应一个逻辑log

topic相当于消息存放目录，partition是topic物理上的分组，一个topic可分为多个partition，每个partition是一个有序的队列

环境搭建：

1）关闭各节点防火墙、selinux

2）各节点安装jdk

3）搭建zookeeper集群环境

4）搭建kafka集群

zookeeper安装包地址：http://archive.apache.org/dist/zookeeper/

节点jdk安装  略

[root@lb ~]# ansible all -a "java -version"
192.168.53.7 | CHANGED | rc=0 >>
openjdk version "1.8.0_161"
OpenJDK Runtime Environment (build 1.8.0_161-b14)
OpenJDK 64-Bit Server VM (build 25.161-b14, mixed mode)
192.168.53.8 | CHANGED | rc=0 >>
openjdk version "1.8.0_161"
OpenJDK Runtime Environment (build 1.8.0_161-b14)
OpenJDK 64-Bit Server VM (build 25.161-b14, mixed mode)
192.168.53.6 | CHANGED | rc=0 >>
openjdk version "1.8.0_161"
OpenJDK Runtime Environment (build 1.8.0_161-b14)
OpenJDK 64-Bit Server VM (build 25.161-b14, mixed mode)

zookeeper集群搭建

下载zookeeper安装包并解压

[root@lb ~]# wget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz
[root@lb ~]# tar xf zookeeper-3.4.10.tar.gz -C /usr/local/
[root@lb local]# ln -s /usr/local/zookeeper-3.4.10/ /usr/local/zookeeper

修改zoo_sample.cfg配置文件

主要修改两个地方

　　1、dataDir=/usr/local/zookeeper/data　　指定完之后data目录要手动创建

　　2、server.0=192.168.53.6:2888:3888

　　　  server.1=192.168.53.7:2888:3888

　　　  server.2=192.168.53.8:2888:3888

[root@lb ~]# cd /usr/local/zookeeper/conf
[root@lb conf]# ls
configuration.xsl  log4j.properties  zoo_sample.cfg
[root@lb conf]# mv zoo_sample.cfg zoo.cfg        //原有的名字并不规范
[root@lb conf]# grep -Ev "#|^$" zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper/data        ##目录是用来存放myid信息跟一些版本，日志，跟服务器唯一的ID信息等
clientPort=2181
server.0=192.168.53.6:2888:3888
server.1=192.168.53.7:2888:3888
server.2=192.168.53.8:2888:3888

说明：

　　clientPort=2181　　客户端连接的接口，客户端连接zookeeper服务器的端口，zookeeper会监听这个端口，接收客户端的请求访问！这个端口默认是2181

　　在zoo.cfg这个文件中，配置集群信息是存在一定的格式　

　　　　server.服务器编号（myid里面的值）:服务器地址:服务端内部通信的端口（默认2888）:选举端口（默认是3888）

创建服务器标识

[root@lb zookeeper]# pwd
/usr/local/zookeeper
[root@lb zookeeper]# mkdir data
[root@lb zookeeper]# echo "0" >data/myid
[root@lb zookeeper]# cat data/myid
0

将zookeeper目录分发各节点，还有环境变量的配置文件

[root@lb zookeeper]# ansible node -a "ln -s /usr/local/zookeeper-3.4.10 /usr/local/zookeeper"

修改节点myid文件

[root@web1 zookeeper]# echo "1" >data/myid
[root@web2 zookeeper]# echo "2" >data/myid

启动zookeeper

用bin下面的脚本zkServer.sh来启停

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

查看状态

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

kafka集群搭建

安装包下载地址：http://mirror.bit.edu.cn/apache/kafka/

下载安装包并解压

[root@lb ~]# wget http://mirror.bit.edu.cn/apache/kafka/2.4.1/kafka_2.12-2.4.1.tgz
[root@lb ~]# tar xf kafka_2.12-2.4.1.tgz -C /usr/local/
[root@lb ~]# ln -s /usr/local/kafka_2.12-2.4.1/ /usr/local/kafka

修改配置文件server.properties

192.168.53.6需修改的地方如下：（其他节点服务器也需做相应的修改）

　　broker.id=0

　　listeners=PLAINTEXT://192.168.53.6:9092

　　zookeeper.connect=192.168.53.6:2181,192.168.53.7:2181,192.168.53.8:2181

　　log.dirs=/usr/local/kafka/logs　　#日志路径 方便查看日志

创建日志的存放路径

　　mkdir -p /usr/local/kafka/logs

将kafka目录发送各节点

[root@lb local]# scp -rp /usr/local/kafka_2.12-2.4.1/ 192.168.53.7:/usr/local/
[root@lb local]# scp -rp /usr/local/kafka_2.12-2.4.1/ 192.168.53.8:/usr/local/
[root@lb local]# ansible node -a "ln -s /usr/local/kafka_2.12-2.4.1 /usr/local/kafka"

分别对分发过去的配置文件进行修改

　　如：53.7　　server.properties

　　broker.id=1

　　listeners=PLAINTEXT://192.168.53.7:9092

启动kafka（kafka/bin/ 目录下有服务启动和停止的脚本）

[root@lb bin]# ./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties  

[root@web1 kafka]# /usr/local/kafka/bin/kafka-server-start.sh -daemon config/server.properties

[root@web2 local]# /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

创建一个topic

创建一个名为test的topic，一个分区一个副本　　（一个节点创建topic之后会自动同步到另外的两个broker上，可在另外两个broker上用--list进行查看）

[root@lb bin]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.53.6:2181 --replication-factor 1 --partitions 1 --topic test

--partitions 1　　指定分区，一个分区表示只会在一个borker进行存放　　如果是3就是3个broker进行均摊存放

运行--list查看这个topic

[root@lb bin]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.53.6:2181
test

[root@web1 kafka]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.53.7:2181
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
test

[root@web2 local]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.53.8:2181
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
test

发送消息

kafka自带命令行客户端，它从文件或者标准输入中读取输入，并将其作为message（消息）发送到集群

[root@lb ~]# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 192.168.53.6:9092 --topic test
>111
>

启动一个consumer 

kafka还有一个命令行consumer（消费者），将消息转储到标准输出

[root@web1 kafka]# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.53.6:9092 --topic test --from-beginning
111

 

 可以看到数据只存放在了主节点，如果分区指定为3个得话，3个节点会分别都存放