zokeeper+kafka

zookeeperkafka

zokeeper+kafka

zokeeper

http://zookeeper.apache.org/
http://mirror.bit.edu.cn/apache/zookeeper
http://zookeeper.apache.org/doc/r3.4.14/zookeeperStarted.html
http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

安装jdk

rpm -ivh jdk-8u221-linux-x64.rpm

安装zookeeper

tar xf zookeeper-3.4.14.tar.gz  -C /usr/local/src/
ln -sv /usr/local/src/zookeeper-3.4.14/ /usr/local/zookeeper


/usr/local/zookeeper/conf
cp zoo_sample.cfg zoo.cfg

grep "^[a-Z]" zoo.cfg
tickTime=2000 #服务器与服务器之间和客户端与服务器之间的单次心跳检测时间间隔，单位是毫秒
initLimit=10 #集群中leader 服务器与follower服务器初始连接心跳次数，即多少个2000毫秒
syncLimit=5  #leader与follower之间连接完成之后，后期检测发送和应答的心跳次数，如果该follower 在设置的时间内（5*2000）不能与leader 进行通信，那么此follower将被视为不可用。
dataDir=/usr/local/zookeeper/data #自定义的zookeeper保存数据的目录。
clientPort=2181#客户端连接zookeeper服务器的端口，Zookeeper会监听这个端口，接受客户端的访问请求
server.1=192.168.10.211:2888:3888 服务器节点 IP/HOST 同步数据 领导者选举
server.2=192.168.10.212:2888:3888
server.3=192.168.10.213:2888:3888

autopurge.snapRetainCount=3 #设置zookeeper保存保留多少次客户端连接的数据
autopurge.purgelnterval=1 #设置zookeeper间隔多少小时清理一次保存的客户端数据

建立数据目录

mkdir -p /usr/local/zookeeper/data
chown 2002.2002 data/

3台尽量同时启动

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

有报错

/usr/local/zookeeper/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Error contacting service. It is probably not running.

查看日志

cat zookeeper.out
at org.apache.zookeeper.server.quorum.QuorumPeerMain.main(QuorumPeerMain.java:81)
Caused by: java.lang.IllegalArgumentException: /usr/local/zookeeper/data/myid file is missing

解决方法

[root@zookeeper1 ~]# echo 1 > /usr/local/zookeeper/data/myid
[root@zookeeper2 ~]# echo 2 > /usr/local/zookeeper/data/myid

[root@zookeeper3 ~]# echo 3 > /usr/local/zookeeper/data/myid
各节点在启动服务

ss -tnl

[root@zookeeper1 ~]# /usr/local/zookeeper/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower
[root@zookeeper2 ~]# /usr/local/zookeeper/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower
[root@zookeeper3 ~]# /usr/local/zookeeper/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: leader

开机自启

tail -n1 /etc/rc.d/rc.local
/usr/local/zookeeper/bin/zkServer.sh start

zookeeper简单操作命令
连接到任意节点生成数据

/usr/local/zookeeper/bin/zkCli.sh  -server 192.168.10.213:2181
[zk: 192.168.10.213:2181(CONNECTED) 0] create /test "hello"
Created /test
[zk: 192.168.10.213:2181(CONNECTED) 1] get /test
hello
cZxid = 0x100000019
ctime = Thu Aug 08 13:49:08 CST 2019
mZxid = 0x100000019
mtime = Thu Aug 08 13:49:08 CST 2019
pZxid = 0x100000019
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 5
numChildren = 0

在其他zookeeper节点验证数据
[zk: 192.168.10.212:2181(CONNECTED) 0]
hello
cZxid = 0x100000019
ctime = Thu Aug 08 13:49:08 CST 2019
mZxid = 0x100000019
mtime = Thu Aug 08 13:49:08 CST 2019
pZxid = 0x100000019
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 5
numChildren = 0

win10下载工具包访问
https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip

cd D:PersonalDownloadsCompressedooInspectoruild>java -jar zookeeper-dev-ZooInspector.jar
d:
java -jar zookeeper-dev-ZooInspector.jar

zookeeper

zookeeper

zookeeper

kafka

http://kafka.apache.org/quickstart
http://kafka.apache.org/downloads
http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz

tar xf kafka_2.12-2.3.0.tgz -C /usr/local/src/
ln -sv /usr/local/src/kafka_2.12-2.3.0/ /usr/local/kafka

修改配置文件

vim /usr/local/kafka/config/server.properties
brokerid=1 #设置每个节点全局唯一的整数ID
listeners=PLAINTEXT://192.168.10.211:9092 
log.retention.hours=168 #保留指定小时的日志内容
zookeeper.connect=192.168.10.211:2181,192.168.10.212:2181,192.168.10.213:2181 #所有zookeeper地址

启动kafka

/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

我试了半天启不来，一直报连接超时

1、时间同步
2、防火墙检查
3、内存
vim /usr/local/kafka/bin/kafka-server-start.sh 
export KAFKA_HEAP_OPTS="-Xmx200m -Xms200m"
4、解析
/etc/hosts
192.168.10.211 zookeeper1 kafka1
192.168.10.212 zookeeper2 kafka2
192.168.10.213 zookeeper3 kafka3

测试kafka
验证进程 jps
[root@zookeeper3 ~]# jps
22699 Jps
15629 QuorumPeerMain
22445 Kafka
查看端口 ss -tnl

测试创建topic
创建名为logstashtest，partitions(分区)为3，replication(复制)为3的topic(主题)：
在任意kafaka服务器操作

[root@zookeeper1 ~]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.10.211:2181 --partitions 3 --replication-factor 3 --topic logstashtest
Created topic logstashtest.

测试获取topic
可以在任意一台kafka服务器进行测试

[root@zookeeper3 ~]# /usr/local/kafka/bin/kafka-topics.sh  --describe --zookeeper 192.168.10.213:2181  --topic logstashtest
Topic:logstashtest	PartitionCount:3	ReplicationFactor:3	Configs:
    Topic: logstashtest	Partition: 0	Leader: 1	Replicas: 1,3,2	Isr: 1,3,2
    Topic: logstashtest	Partition: 1	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3
    Topic: logstashtest	Partition: 2	Leader: 3	Replicas: 3,2,1	Isr: 3,2,1

状态说明：logstashtest有三个分区分别为0、1、2，分区0的leader是3（broker.id），分区0有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）。

删除topic

/usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 192.168.10.213:2181  --topic logstashtest

获取所有topic

/usr/local/kafka/bin/kafka-topics.sh  --list --zookeeper 192.168.10.211:2181

kafka命令测试消息发送

创建topic

[root@zookeeper3 ~]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.10.211:2181 --partitions 3 --replication-factor 3 --topic  messagetest
Created topic messagetest.

发送消息

[root@zookeeper3 ~]# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list  192.168.10.211:9092 --topic  messagetest
>hello
>test
>test2
>bye   
>

其他kafka服务器测试获取数据

[root@zookeeper1 ~]# /usr/local/kafka/bin/kafka-console-consumer.sh  --topic messagetest  --bootstrap-server 192.168.10.211:9092 --from-beginning
test2
hello
bye
test