Kafka Connect MQTT 项目教程

2024-08-15 09:59:47作者：滑思眉Philip

1. 项目的目录结构及介绍

kafka-connect-mqtt/
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   ├── be/
│   │   │   │   ├── jovacon/
│   │   │   │   │   ├── kafka/
│   │   │   │   │   │   ├── connect/
│   │   │   │   │   │   │   ├── mqtt/
│   │   │   │   │   │   │   │   ├── MQTTSinkConnector.java
│   │   │   │   │   │   │   │   ├── MQTTSourceConnector.java
│   │   │   │   │   │   │   │   ├── MQTTSinkTask.java
│   │   │   │   │   │   │   │   ├── MQTTSourceTask.java
│   │   │   │   │   │   │   │   ├── MQTTConnectorConfig.java
│   │   │   │   │   │   │   │   ├── MQTTMessageConverter.java
│   │   │   │   │   │   │   │   ├── MQTTUtils.java
├── pom.xml
├── README.md

src/main/java/be/jovacon/kafka/connect/mqtt/：包含 MQTT 连接器的核心代码，包括源和接收连接器的实现。
pom.xml：Maven 项目配置文件，用于构建项目。
README.md：项目说明文档，包含项目的基本信息和使用指南。

2. 项目的启动文件介绍

MQTTSinkConnector.java：MQTT 接收连接器的主要类，负责将 Kafka 消息发送到 MQTT 主题。
MQTTSourceConnector.java：MQTT 源连接器的主要类，负责从 MQTT 主题订阅消息并写入 Kafka 主题。
MQTTSinkTask.java 和 MQTTSourceTask.java：分别是接收和源任务的实现，处理具体的消息传输任务。

3. 项目的配置文件介绍

`pom.xml`

Maven 项目配置文件，定义了项目的依赖、构建配置等。

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>be.jovacon</groupId>
    <artifactId>kafka-connect-mqtt</artifactId>
    <version>1.0.0</version>
    <dependencies>
        <!-- 依赖项列表 -->
    </dependencies>
    <build>
        <plugins>
            <!-- 构建插件列表 -->
        </plugins>
    </build>
</project>

`README.md`

项目说明文档，包含项目的基本信息和使用指南。

# Kafka Connect MQTT

## 简介
Kafka Connect MQTT 是一个用于 Apache Kafka 的 MQTT 连接器，支持 MQTT 源和接收连接器。

## 安装
1. 克隆仓库：
   ```sh
   git clone https://github.com/evokly/kafka-connect-mqtt.git

构建项目：
```
mvn clean install
```
将生成的连接器文件复制到 Kafka Connect 插件路径并重启 Kafka Connect。

配置

源连接器配置示例

{
  "name": "mqtt-source-connector",
  "config": {
    "connector.class": "be.jovacon.kafka.connect.mqtt.MQTTSourceConnector",
    "mqtt.topic": "my_mqtt_topic",
    "kafka.topic": "my_kafka_topic",
    "mqtt.clientID": "my_client_id",
    "mqtt.broker": "tcp://127.0.0.1:1883",