一、我的服务器环境:CentOS7,kafka依赖JVM环境,要求安装有JDK
使用CentOS7系统,先删除自带的JDK,自带的不好用(例如jps终端命令不能用等)。下载官网完整的JDK,并设置环境变量JAVA_HOME。详情见:
https://blog.csdn.net/libaineu2004/article/details/80060812
[root@localhost ~]# java -version
openjdk version "1.8.0_141"
OpenJDK Runtime Environment (build 1.8.0_141-b16)
OpenJDK 64-Bit Server VM (build 25.141-b16, mixed mode)
二、下载
Kafka(http://kafka.apache.org/)依赖ZooKeeper
http://mirrors.shuosc.org/apache/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz --- 稳定版
http://mirrors.shuosc.org/apache/kafka/1.1.0/kafka-1.1.0-src.tgz
http://mirror.bit.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz
由于kafka_2.12-1.1.0.tgz是可执行程序,里面集成了zookeeper简易版,如果只是单机测试那么仅仅下载这个就可以了。但是简易版功能有限。在生产环境下,建议还是直接下载官方zookeeper软件。
三、运行与测试,以下都是终端命令
1、cd kafka_2.12-1.1.0
2、启动Zookeeper,如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动,即守护进程。
方法1(推荐) ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
方法2 nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties&
3、启动Kafka,如果无报错则说明启动成功。daemon以及nohup &是实现在后台启动,即守护进程。
方法1 (推荐)./bin/kafka-server-start.sh -daemon config/server.properties
方法2 nohup ./bin/kafka-server-start.sh config/server.properties&
4、查看进程是否正常运行,通过检测2181与9092端口,2181是zookeeper,9092是kafka
netstat -tunlp|egrep "(2181|9092)"
tcp 0 0 :::2181 :::* LISTEN 19787/java
tcp 0 0 :::9092 :::* LISTEN 28094/java
我们也可以使用jps终端命令查看是否启动:
[root@bogon kafka_2.12-1.1.0]# jps
4384 Kafka
3921 QuorumPeerMain
4412 Jps
5、创建主题,例如test
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --partitions 3 --topic emqtest
--zookeeper : zookeeper集群列表,用英文逗号分隔。可以不用指定zookeeper整个集群内的节点列表,只指定某个或某几个zookeeper节点列表也是可以的
replication-factor : 复制数目,提供failover机制;1代表只在一个broker上有数据记录,一般值都大于1,代表一份数据会自动同步到其他的多个broker,防止某个broker宕机后数据丢失。
partitions : 一个topic可以被切分成多个partitions,一个消费者可以消费多个partitions,但一个partitions只能被一个消费者消费,所以增加partitions可以增加消费者的吞吐量。kafka只保证一个partitions内的消息是有序的,多个一个partitions之间的数据是无序的。
注意,
(1)使用集群时,必须手动先创建topic,指定zookeeper节点列表;单机也应该要创建。否则客户端会报错。
(2)主题名不建议使用"."和"_"字符。
[root@localhost kafka_2.12-1.0.0copy1]# ./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --
partitions 3 --topic emq_broker_message1
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic "emq_broker_message1".
[root@localhost kafka_2.12-1.0.0copy1]# ./bin/kafka-topics.sh --create --zookeeper 172.16.6.170:2181,172.16.6.170:2182,172.16.6.170:2183 --replication-factor 3 --
partitions 3 --topic emqtest
Created topic "emqtest".
(3)建议创建主题时,指定分区数为kafka broker节点数目的整数倍,例如broker有3台,那么分区就设置为3个或6个或9个--partitions 3。创建完主题,可以在路径log.dirs看到主题和分区的文件夹。详情参见《Kafka入门与实践.牟大恩》3.6.1节描述
(4)--replication-factor 3,《kafka权威指南》6.3.1节建议复制系数为3.但是前提是数值不能大于broker的节点数。
如果broker节点数为1,复制系数设置为3,则会报错:
Error while executing topic command : Replication factor: 3 larger than available brokers: 1.
(5)单机时,zookeeper数目是1;集群时,zookeeper的数目建议为奇数个,至少3个。
(6)server.properties文件设置默认值
num.partitions=3
default.replication.factor=1
6、查看主题
./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
./bin/kafka-topics.sh --list --zookeeper localhost:2181
例如:broker集群,有3个节点。同时给主题mynewt分配了3个分区。每个分区都有leader和follow。
[root@bogon kafka_2.12-1.1.0]# ./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic mynewt
Topic:mynewt PartitionCount:3 ReplicationFactor:3 Configs:
Topic: mynewt Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: mynewt Partition: 1 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: mynewt Partition: 2 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
7、开启一个终端,发送消息,生产者的消息要发往kafka
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
8、另起一个终端,消费消息,消费者的消息来自zookeeper(协调转发)
./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
将来新版本需要写成./bin/kafka-console-consumer.sh --bootstrap-server localhost:2181 --topic test --from-beginning
9、结束进程
./bin/kafka-server-stop.sh
./bin/zookeeper-server-stop.sh
kafka启动时先启动zookeeper,再启动kafka;关闭时相反,先关闭kafka,再关闭zookeeper。
四、总结
1、kafka有什么?
producer 消息的生成者,即发布消息
consumer 消息的消费者,即订阅消息
broker Kafka以集群的方式运行,可以由一个或多个服务组成,服务即broker
zookeeper 协调转发
2、配置文件与消息持久化路径
/config/server.properties
# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
例如客户端发送test为主题的消息,会持久化在这个文件:/tmp/kafka-logs/test-0/00000000000000000000.log
主题列表会记录在/tmp/zookeeper
/config/zookeeper.properties
# the directory where the snapshot is stored.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
建议修改kafka的日志目录和zookeeper数据目录,因为这两项默认放在tmp目录,而tmp目录中内容会随重启而丢失:
server.properties:
log.dirs=/tmp/kafka-logs
修改为
log.dirs=/usr/local/kafka/logs
zookeeper.properties
dataDir=/tmp/zookeeper
修改为
dataDir=/usr/local/zookeeper/data
3、配置文件bootstrap.servers 和 broker.list区别:
本以为是两个参数,其实是实现一个功能,查看源代码后发现broker.list是旧版本命令
4、自己写的shell启动和关闭kafka脚本,kafka.sh -- 这个脚本有问题,不稳定
#!/bin/bash kafka_home=/root/Downloads/kafka_2.12-1.0.0/ case $1 in start) # 服务启动需要做的步骤 echo "zookeeper start" $kafka_home/bin/zookeeper-server-start.sh -daemon $kafka_home/config/zookeeper.properties #sleep 1 echo "kafka start" $kafka_home/bin/kafka-server-start.sh -daemon $kafka_home/config/server.properties #sleep 1 ;; stop) # 服务停止需要做的步骤 echo "kafka stop" $kafka_home/bin/kafka-server-stop.sh #sleep 1 echo "zookeeper stop" $kafka_home/bin/zookeeper-server-stop.sh #sleep 1 ;; restart) # 重启服务需要做的步骤 ... ;; status) # 查看状态需要做的步骤 ... ;; *) echo "$0 {start|stop|restart|status}" exit 4 ;; esac
5、zookeeper集群
kafka是通过zookeeper来管理集群。 kafka软件包内虽然包括了一个简版的zookeeper,但是感觉功能有限。在生产环境下,建议还是直接下载官方zookeeper软件。 下载最新版的zookeeper软件:
http://mirror.bit.edu.cn/apache/zookeeper/
zookeeper-3.4.11.tar.gz
欢迎访问姊妹篇:我个人的kafka broker和zookeeper集群实践(★firecat推荐★)
---
参考文献
物联网架构成长之路(8)-EMQ-Hook了解、连接Kafka发送消息
centos7 安装 kafka_2.11-1.0.0
zookeeper安装及部署
Kafka安装及部署
CentOS下ZooKeeper单机模式、集群模式安装
Kafka单机、集群模式安装详解(一)
kafka集群监控之KafkaOffsetMonitor 0.4.1版本 以及用sbt编译KafkaOffsetMonitor
https://github.com/quantifind/KafkaOffsetMonitor