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

数据流在当今的数据环境中，没有一个系统可以提供所有必需的观点来提供真正的洞察力。从数据中获取完整含义需要混合来自多个来源的大量信息。
与此同时，我们不耐烦地立即获得答案;如果洞察时间超过10毫秒，那么该值就会丢失 -
高频交易，欺诈检测和推荐引擎等应用程序不能等待。这通常意味着在数据进入记录数据库之前分析数据的流入。为数据丢失增加零容忍，挑战变得更加艰巨。
Kafka和数据流专注于从多个消防软管摄取大量数据，然后将其路由到需要它的系统 - 过滤，汇总和分析途中。本文介绍了Apache
Kafka，然后演示了如何使用MongoDB作为流数据的源（生产者）和目标（消费者）。有关此主题的更完整的研究可以在使用Kafka和MongoDB白皮书的Data
Streaming中找到。

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主题中。

公众号推荐：

公众号：VOA英语每日一听

微信号: voahk01

可长按扫码关注，谢谢

热门工具换一换