通过 MQTT 将 PLCnext Control 无缝接入 Apache Kafka：企业级集成方案

技术背景

Apache Kafka

Apache Kafka 是用于数据采集、存储、处理和再分发的框架，现已在全球企业广泛部署。其官网提供更多理念与部署信息。核心特性之一是拥有海量现成连接器，支持 MQTT 等协议与应用集成。

MQTT

MQTT 是基于 TCP 的轻量级消息协议，因其健壮性和小体量，常用于物联网通信。OASIS 标准详情见其官网。

此前有文章介绍如何为 PLCnext 交叉编译 Eclipse Mosquitto，PLCnext Store 也提供现成 MQTT 应用。

需求清单

• PLCnext 上的 MQTT 客户端（实现提示见上文）
• 控制器与 PC/VM 网络互通
• PC/VM 上运行 MQTT Broker（如 Mosquitto）
• PC/VM 上运行 Kafka 实例（参考 Kafka 快速入门指南）

架构搭建

下图展示将 PLCnext 数据导入 Kafka 的架构概览。虽可使用 Confluent MQTT Proxy（方案 2），但本文聚焦更通用的方案 1：MQTT 客户端连接 Broker 发布消息，连接器订阅 Broker 主题、处理消息并转发至 Kafka。

创建连接器

本教程基于 GitHub 上 evokly/kafka-connect-mqtt 仓库（MIT 许可证）。首先下载解压仓库。因最新版本停留在 2016 年末，需更新 build.gradle 依赖版本：

ext { kafkaVersion = '2.6.0' }
...
dependencies {
  testCompile group: 'junit', name: 'junit', version: '4.13'
  compile "org.apache.kafka:connect-api:$kafkaVersion"
  compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
  compile 'org.bouncycastle:bcprov-jdk15on:1.67'
  compile 'org.bouncycastle:bcpkix-jdk15on:1.67'
  compile 'org.bouncycastle:bcpg-jdk15on:1.67'
  compile 'commons-io:commons-io:2.8.0'
  compile 'org.slf4j:slf4j-api:1.7.30'
  testCompile 'org.slf4j:slf4j-simple:1.7.30'
}

本例发送纯字符串消息到 Kafka，需修改默认消息处理器 DumbProcessor.java（位于 /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt）：

@Override
public SourceRecord[] getRecords(String kafkaTopic) {
  return new SourceRecord[]{
    new SourceRecord(
      null, null, kafkaTopic, null, null,
      mTopic, null, mMessage.toString(), new Long(123L)
    )
  };
}

随后构建 JAR：./gradlew clean jar。将输出的 kafka-connect-mqtt-1.1-SNAPSHOT.jar（在 /build/libs）复制到 Kafka 的 libs 目录。同时需将 org.eclipse.paho.client.mqttv3-1.2.5.jar 也放入 Kafka 的 libs 目录（下载地址见文中链接）。

在 Kafka 的 config 目录创建连接器配置文件 mqtt.properties：

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
kafka.topic=test_in
mqtt.client_id=mqtt-kafka-123
mqtt.clean_session=true
mqtt.connection_timeout=30
mqtt.keep_alive_interval=60
mqtt.server_uris=tcp://172.17.0.1:1883
mqtt.topic=test/#

本地测试

进入 Kafka 目录，依次启动 ZooKeeper、Kafka Broker、MQTT Broker（Docker 运行 Mosquitto）、MQTT-Kafka 连接器、Kafka 控制台消费者，最后发布测试消息：

# 启动 ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
# 启动 Kafka
bin/kafka-server-start.sh config/server.properties
# 启动 MQTT Broker (Mosquitto Docker)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto
# 启动连接器
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties
# 启动控制台消费者
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true
# 发布测试消息
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

消息将出现在控制台消费者中，验证集成成功。