从 Kafka 消费消息到 EMQ X
搭建 Kafka 环境,以 MaxOS X 为例:
$ wget http://apache.claz.org/kafka/2.3.1/kafka_2.12-2.3.1.tgz
$ tar -xzf kafka_2.12-2.3.1.tgz
$ cd kafka_2.12-2.3.1
# 启动 Zookeeper
$ ./bin/zookeeper-server-start.sh config/zookeeper.properties
# 启动 Kafka
$ ./bin/kafka-server-start.sh config/server.properties
Kafka消费组不支持Kafka0.9以下版本
创建资源之前,需要提前创建Kafka主题,不然会提示错误
创建 Kafka 的主题:
$ ./bin/kafka-topics.sh --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic --create
创建资源:
打开 EMQ X Dashboard,选择左侧的 “资源” 选项卡。
点击 “新建” 按钮:
选择资源类型 “Kafka 消费组”:
填写资源参数:
1). Kafka 服务器地址
2). Kafka consumer 连接池大小
3). Kafka 的订阅主题
4). Kafka 的消息主题
5). Kafka Max Bytes (每次从 Kafka 里消费消息的最大字节数)
6). Kafka Offset Reset Policy (重置Offset策略,reset_to_latest | reset_by_subdcriber)
7). Kafka consumer 是否重连
最后点击 “确认”,资源创建完成:
资源已经创建完成,现在用Dashboard的websocket工具订阅MQTT的主题 “TestTopic”:
使用kafka 命令行 生产一条消息:
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TestTopic
Dashboard的websocket工具接收到了Kafka 生产的消息”hello-kafka”: