环境搭建

为了配合阿里云消息队列Kafka版本，本次使用 `kafka_2.12-0.10.2.2` 版本搭建

Step 1 下载代码

> wget https://archive.apache.org/dist/kafka/0.10.2.2/kafka_2.12-0.10.2.2.tgz

> tar -zxf kafka_2.12-0.10.2.2.tgz

> cd kafka_2.12-0.10.2.2

1. kafka官网下载 ，http://kafka.apache.org/downloads；

2. 找到0.10.2.2版本 http://kafka.apache.org/downloads#0.10.2.2；

3. 下载二进制文件 Binary downloads 地址：https://archive.apache.org/dist/kafka/0.10.2.2/kafka_2.12-0.10.2.2.tgz；

4. 解压文件 tar -xzf kafka_2.12-0.10.2.2.tgz；

5. 进入kafka文件目录 cd kafka_2.12-0.10.2.2；

Step 2: 启动服务

kafka运行需要依赖Zookeeper服务，可以自行搭建zookeeper或直接使用kafka自带打包和配置好的zookeeper

1. 启动zookeeper

nohup bin/zookeeper-server-start.sh config/zookeeper.properties > logs/zookeeper.log &

2. 启动kafka

nohup bin/kafka-server-start.sh config/server.properties > logs/kafka.log &

Step 3: 创建主题(topic)

1. 创建一个test topic，或者，除了手工创建topic外，你也可以配置你的broker，当发布一个不存在的topic时自动创建topic。

bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test

2. 创建好之后查看topic

> bin/kafka-topics.sh --list --zookeeper localhost:2181

test

Step 4: 发送消息

Kafka提供了一个命令行的工具，可以从输入文件或者命令行中读取消息并发送给Kafka集群。每一行是一条消息。

运行producer（生产者）,然后在控制台输入几条消息到服务器。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a message

This is another message

Step 5: 消费消息

如果有2台不同的终端上运行上述命令，那么当你在运行生产者时，消费者就能消费到生产者发送的消息。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

This is a message

This is another message

Step 6: SpringCloud Stream-Kafka

https://github.com/WeekiXu/spring-stream-demo

搭建完成

Step -1: 常见异常

1. 创建消息后打开consumer抛出异常，或producer及consumer无法远程访问

[2019-09-24 11:43:23,308] WARN Error while fetching metadata with correlation id 2 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

[2019-09-24 11:43:23,435] WARN Error while fetching metadata with correlation id 3 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

解决方案

检查防火墙是否开启，如需要开启则开放`9092`端口，参考防火墙配置链接

https://blog.weeki.top/linux/2018/06/20/firewalld/

获取kafka当前部署机器ip及hostname，

例：ip:192.168.3.29，hostname：kafka-local，

将ip对应hostname配置在远程终端的hosts表中

192.168.3.29  kafka-local

添加如下内容到server.properties，重启kafka及zookeeper服务

isteners=PLAINTEXT://0.0.0.0:9092

advertised.port = 9092

advertised.host.name=kafka-local

2. SpringCloud Stream Kafka生产及消费异常

java.lang.IllegalArgumentException: Magic v1 does not support record headers

这是client与broker（server）出现版本不一致的问题。

由于当前版本为配合阿里云Kafka服务基础版，选定为`0.10.2.2`，

SpringBoot版本选定 2.1.1.RELEASE, SpringCloud版本 Greenwich.SR2，

默认依赖的Spring-Kafka版本就是1.0.x及以上了。

但是Broker却是0.10.2版本，所以也能很明显的知道他们进行了版本号校验，

只要把kafka server，即broker版本升级为1.0.x及以上的版本即可解决。

参考文献

kafka单机环境搭建及其基本使用 https://www.cnblogs.com/qpf1/p/9161742.html

https://blog.verysu.com/article/408