前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun
前言
本文讲如何安装启动kafka,并进行测试，其中zookeepr是kafka自带的，本文基本按照官网文档进行安装启动的，并提出可能会出现的问题。官方文档：http://kafka.apache.org/quickstart
本文虚拟机系统：centos7，不过其他版本的Linux系统是一样的~
1、下载
可直接在官网下载对应的版本http://kafka.apache.org/downloads，我下载的是二进制版的，由于我的scala版本是2.11，所以下载kafka_2.11-1.1.0.tgz，大家可以根据自己的实际情况选择对应的版本。执行以下命令即可下载到本地了。
1 | wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz |
2、解压到指定目录
1 | tar -xzf kafka_2.11-1.1.0.tgz -C /opt/ |
3、启动服务
3.1 启动zookeeper
kafka用到zookeeper，因此如果您的机器上没有zookeeper服务，则需要先启动zookpeer服务，本文使用kafka自带的zookeeper。1
2cd /opt/kafka_2.11-1.1.0/
bin/zookeeper-server-start.sh config/zookeeper.properties
因日志较多，所以只给出前几行信息和最后一行成功的信息1
2
3
4[2018-05-21 12:17:44,461] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2018-05-21 12:17:44,472] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
...
[2018-05-21 12:17:44,844] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
可以看到zookeeper服务的端口为2181
3.2 启动kafka服务
打开第二个终端1
bin/kafka-server-start.sh config/server.properties
1 | [2018-05-21 12:21:07,901] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$) |
其中kafka的端口为9092,在下面这条信息可以看到1
INFO Registered broker 0 at path /brokers/ids/0 with addresses: ArrayBuffer(EndPoint(ambari.master.com,9092,ListenerName(PLAINTEXT),PLAINTEXT)) (kafka.zk.KafkaZkClient)
4、创建一个主题
打开第三个个终端1
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
1 | Created topic "test". |
可以通过list topic命令查看所有的主题1
bin/kafka-topics.sh --list --zookeeper localhost:2181
1 | test |
或者，您也可以将代理配置为在发布不存在的主题时自动创建主题，而不是手动创建主题。
5、发送消息
Kafka带有一个命令行客户端，它将从文件或标准输入中获取输入，并将其作为消息发送到Kafka集群。默认情况下，每行将作为单独的消息发送。
启动生产者1
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
输入几条消息发送到服务器1
2>This is a message
>This is another message
6、启动消费者
消费者可以将消息转储到标准输出
打开第四个个终端1
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
然后就可以在命令行看到生产者发送的消息了1
2This is a message
This is another message
7、spark远程连接kafka
因为我是用spark进行开发的，所以此时就想测试一下，这样配置，用spark程序在celipse里是否可以获取到kafka的数据（我之前使用ambari搭建的kafka进行测试的），程序可参考Spark Streaming连接Kafka入门教程,但是运行程序发现并不能获取到kafka对应topic里的消息。
几次尝试，发现修改下面的配置即可1
vim config/server.properties
1 | listeners=PLAINTEXT://192.168.44.129:9092 |
其中kafka默认的端口就是9092，不同的地方是端口前面加上了ip（192.168.44.129）,我想默认的是localhost，spark程序远程通过ip地址和localhost对应不上，所以获取不到kafka的消息，注意在第8部分：配置多个broker的集群，为了和官方文档一致，并没有加上ip，大家可根据需要自行修改。
这样的话就可以在程序里获取到历史消息了，但是如果想新产生几条数据的话，启动命令需要将localhost改为192.168.44.129，否则会产生异常，即：
1 | bin/kafka-console-producer.sh --broker-list 192.168.44.129:9092 --topic test |
8、设置多个broker的集群
到目前为止，我们设置的是单个broker，这样并不好，下面我们还是在这一台机器上设置三个节点。
8.1 为每个broker创建一个配置文件
1 | cp config/server.properties config/server-1.properties |
然后用vim修改
config/server-1.properties:1
2
3broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/tmp/kafka-logs-1
config/server-2.properties:1
2
3broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/tmp/kafka-logs-2
（其中listeners需要把前面的注释也就是#去掉）
broker.id是集群中每个节点唯一且永久的名称，因为我们实在同一个机器上运行这些文件，所以为了避免端口冲突和数据彼此覆盖，我们必须重写它的端口和日志目录。
8.2 启动新节点
我们已经启动了一个节点了（broker.id=0），现在启动两个新节点1
2bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &
8.3 创建三个副本的新主题：
1 | bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic |
运行describe topics查看主题的信息1
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
1 | Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: |
下面解释一下这些输出，第一行是所有分区的摘要，另外的每一行是一个分区的信息，因为这个主题只有一个分区，所以只有一行。
leader：负责所有读和写，是这个分区从所有节点随机选择的。
replicas：是为这个分区复制日志的节点列表，无论他们是领导者还是他们现在还活着。
isr：是同步副本的集合，是还活着的副本的自己并被leader捕获（caught-up）。
在我的示例中节点0是该主题唯一分区的leader。
我们可以用相同的命令查看之前创建的主题 test1
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
1 | Topic:test PartitionCount:1 ReplicationFactor:1 Configs: |
8.4 发送消息
1 | bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic |
1 | >my test message 1 |
8、5 消费消息
1 | bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic |
1 | my test message 1 |
8.6 测试容错及发现的问题
我的leader是节点0，现在kill掉
1 | ps aux | grep server.properties |
得到对应的进程号
1 | root 16557 34.6 8.4 4322904 326472 pts/1 Sl+ 05:30 ... |
然后kill掉第一个（第二个是grep命令本身的）1
kill -9 16557
kill掉之后再描述下topic1
2
3bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 0,1,2 Isr: 1,2
现在leader已经切换到节点1，Isr也只有节点1和2了
官网上说：但是即使原来的leader失败，这些消息仍然可用于消费，但是下面又有一些坑
首先执行下面这句1
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
发现会报错1
[2018-05-22 03:55:13,304] WARN [Consumer clientId=consumer-1, groupId=console-consumer-29320] Connection to node -1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
原因是节点0已经kill掉了，也就是端口9092已经不能用了，要换成别的端口，所以把上面的命令的端口换成节点1或者2，再执行，发现不报错了，但是没有得任何数据，即使再生成几条消息，也是不报错，但没有数据。
于是想是不是因为节点0没有加 &，加上 &重新启动节点0，新建主题，直到找到leader为节点0的，再kill掉节点0发现还是同样的问题，但是官网上说是可以的…
找一个leader不是节点0的主题进行测试，比如官网例子上的节点1，kill掉节点1，之前的leader由1变为0了，再消费测试，发现不管是新增的消息还是历史消息都可以得到，至于这是不是官网的bug及如何解决这个bug（可能还需要修改其他配置），目前我还没找到，如果找到我会更新的。
本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。
非商业转载请注明作者及出处。商业转载请联系作者本人。
本文标题：Kafka安装启动入门教程
本文链接：https://dongkelun.com/2018/05/21/kafkaConf/
欢迎关注我的公众号