【MongoDB 新搭档 Kafka】

在MongoDB中，我们可以使用MongoDB Connector for Kafka来将MongoDB中的数据变化实时同步到Kafka中。以下是一个使用MongoDB Connector for Kafka的基本示例。

首先，确保你已经安装了MongoDB和Kafka。

1. 下载并安装MongoDB Connector for Kafka。你可以从https://www.confluent.io/hub/mongodb/kafka-connect-mongodb下载。
2. 配置MongoDB Connector for Kafka。创建一个配置文件mongodb-source-quickstart.properties，如下所示：

name=mongodb-source-quickstart
connector.class=com.mongodb.kafka.connect.MongoDbSourceConnector
tasks.max=1
mongodb.hosts=localhost:27017
mongodb.name=your_mongodb_database
mongodb.user=your_mongodb_user (if applicable)
mongodb.password=your_mongodb_password (if applicable)
topic.prefix=mongodb_source_quickstart_

3. 配置Kafka Connector。创建另一个配置文件kafka-sink-quickstart.properties，如下所示：

name=kafka-sink-quickstart
connector.class=io.confluent.connect.kafka.KafkaSinkConnector
tasks.max=1
topics=mongodb_source_quickstart_your_topic
kafka.topic=your_kafka_topic
bootstrap.servers=localhost:9092

4. 启动MongoDB Connector for Kafka。

./bin/connect-standalone ./etc/kafka-connect-mongodb.properties ./etc/mongodb-source-quickstart.properties

5. 启动Kafka Connector。

./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic your_kafka_topic < ./etc/kafka-sink-quickstart.properties

现在，每当MongoDB数据库中的your_mongodb_database数据库的your_topic主题发生变化时，这些变化都会被实时地同步到Kafka的your_kafka_topic主题中。

注意：以上代码是基于假设你的MongoDB和Kafka运行在本地机器上且没有设置用户认证。根据你的实际环境，你可能需要调整mongodb.hosts, bootstrap.servers, mongodb.name, mongodb.user, mongodb.password, topic.prefix, 和kafka.topic等参数。