MongoDB和数据流：使用MongoDB作为Kafka消费者

Apache Kafka

Kafka提供了一种灵活，可扩展且可靠的方法，用于将来自一个或多个生产者的事件数据流传达给一个或多个消费者。事件的例子包括：

定期传感器读数，例如当前温度 用户在网上商店中将商品添加到购物车中 正在发送带有特定主题标签的Tweet

Kafka事件流被组织成主题。生产者选择一个主题来发送给定的事件，而消费者则选择他们从哪个主题中提取事件。例如，金融应用程序可以从一个主题中提取纽约证券交易所股票交易，并从另一个主题中提取公司财务公告，以寻找交易机会。

在Kafka中，话题被进一步分成多个分区来支持扩展。每个Kafka节点（代理）负责接收，存储和传递来自一个或多个分区的针对给定主题的所有事件。这样，一个主题的处理和存储可以在许多Broker中线性扩展。类似地，应用程序可以通过针对给定主题使用许多消费者来扩展，每个拉事件来自离散的一组分区。

图1：Kafka生产者，消费者，主题和分区

MongoDB作为Kafka消费者的一个Java示例 为了将MongoDB作为Kafka消费者使用，接收到的事件必须先转换为BSON文档，然后再存储到数据库中。在这个例子中，事件是代表JSON文档的字符串。这些字符串被转换为Java对象，以便Java开发人员可以轻松使用;那些对象然后被转换成BSON文档。

完整的源代码，Maven配置和测试数据可以在下面找到，但这里有一些亮点;从用于接收和处理来自Kafka主题的事件消息的主循环开始：

Fish类包含辅助方法以隐藏对象如何转换为BSON文档：

在实际的应用程序中，接收到的消息可能会更多 - 它们可以与从MongoDB读取的参考数据结合使用，然后通过发布到其他主题来处理并传递。在这个例子中，最后一步是从mongo shell确认数据已经添加到数据库中：

MongoDB Kafka Consumer的完整Java代码 业务对象 - Fish.java

MongoDB的Kafka使用者 - MongoDBSimpleConsumer.java

请注意，此示例消费者是使用Kafka Simple Consumer API编写的 - 还有一个Kafka高级消费者API，它隐藏了很多复杂性 - 包括管理偏移量。 Simple API为应用程序提供了更多控制权，但需要花费额外的代码。

Maven依赖- pom.xml

测试数据 - Fish.json注入Kafka的测试数据示例如下所示：

对于简单测试，可以使用kafka-console-producer.sh命令将此数据注入到clusterdb-topic1主题中。