Интеграция PLCnext с Apache Kafka через MQTT: Enterprise-решение

Технический фон

Kafka

Apache Kafka — фреймворк для приёма, хранения, обработки и перераспределения данных. Широко используется во всем мире. Ключевая особенность — множество готовых коннекторов к другим приложениям и протоколам, включая MQTT.

MQTT

MQTT — лёгкий протокол обмена сообщениями на базе TCP, популярный в IoT благодаря надёжности и малому размеру. Стандарт OASIS. Для PLCnext можно скомпилировать mosquitto (Eclipse) или использовать готовые приложения из PLCnext Store.

Требования

• MQTT-клиент на PLCnext
• Контроллер подключён к ПК/VM
• MQTT-брокер на ПК/VM (например, mosquitto)
• Экземпляр Kafka на ПК/VM (руководство по быстрому старту Kafka)

Схема подключения

На схеме показан обзор реализации для передачи данных из PLCnext в Kafka. Возможно использование Confluent MQTT Proxy (2), но рассматривается более универсальное решение (1): MQTT-брокер, к которому подключается клиент, и коннектор, подписывающийся на топик, обрабатывающий сообщения и отправляющий их в Kafka.

Создание коннектора

Коннектор основан на репозитории evokly/kafka-connect-mqtt (MIT License). Скачайте и распакуйте. Обновите 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'
}

Для отправки простых строковых сообщений отредактируйте класс 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 в директорию libs Kafka. Также нужен org.eclipse.paho.client.mqttv3-1.2.5.jar (скачать здесь).

Создайте конфигурационный файл mqtt.properties в папке config Kafka:

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/#
# message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Локальный тест

Запустите ZooKeeper, Kafka, MQTT-брокер (mosquitto в Docker), коннектор и консольного потребителя. Опубликуйте тестовое сообщение — оно появится в консоли.

# start ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
# start Kafka
bin/kafka-server-start.sh config/server.properties
# start MQTT broker
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto
# start connector
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties
# start consumer
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true
# publish message
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123