那间公司做网站好,成都网站优化师,ui在线设计平台,百度推广助手MongoDB是一个非常出色的“ NoSQL”数据库#xff0c;具有广泛的应用程序。 在SoftwareMill开发的一个项目中#xff0c;我们将其用作复制的事件存储#xff0c;然后将事件从事件流传输到其他组件。 介绍 基本思想非常简单#xff08;另请参阅Martin Fowler关于Event Sou… MongoDB是一个非常出色的“ NoSQL”数据库具有广泛的应用程序。 在SoftwareMill开发的一个项目中我们将其用作复制的事件存储然后将事件从事件流传输到其他组件。 介绍 基本思想非常简单另请参阅Martin Fowler关于Event Sourcing的文章。 我们的系统生成一系列事件。 这些事件将保留在事件存储中。 系统中的其他组件遵循事件流并对其进行“处理”。 例如可以将它们汇总并写入报告数据库另一方面它类似于CQRS 。 这种方法有很多优点 事件的读取和写入是解耦的异步的 鉴于它没有死得太久任何后续组件都可能死亡然后“追赶” 可能有多个关注者。 跟随者可以从从属副本读取数据以获得更好的可伸缩性 事件活动的爆发对事件接收器的影响减少 最坏的情况下报告生成速度会变慢 这里的关键组件当然是快速可靠的事件存储。 我们用来实现一个的MongoDB的三个关键功能是 上限集合和尾部游标 快速收集附件 复制集 采集 作为基础我们使用有上限的集合 根据定义该集合受大小限制。 如果编写新事件将导致集合超出大小限制则最早的事件将被覆盖。 这给了我们类似于事件的循环缓冲区的功能。 此外我们也很安全地避免了磁盘空间不足错误。 在2.2版之前默认情况下上限集合没有_id字段因此没有索引。 但是由于我们希望事件能够在整个副本集上可靠地写入因此_id字段及其上的索引都是必需的。 写作活动 编写事件是一个简单的Mongo插入操作 插入也可以分批完成。 根据我们对事件丢失的容忍度我们可能会使用各种Mongo 写入问题 例如等待来自单节点或多个节点的写入确认。 所有事件都是不可变的。 除了更好的线程安全的Java代码外这是事件流的必要条件。 如果事件是可变的事件接收器将如何知道更新的内容 而且这对Mongo的性能有很好的影响。 由于永远不会更改数据因此写入磁盘的文档永远不会缩小或扩展因此无需在磁盘上移动块。 实际上在具有上限的集合中Mongo不允许增长曾经编写的文档。 阅读活动 读取事件流要复杂一些。 首先可能有多个阅读器每个阅读器在流中具有不同的进度。 其次如果流中没有事件我们希望读者等待一些事件可用并避免主动轮询。 最后我们想分批处理事件以提高性能。 有尾游标可以解决这些问题。 要创建这样的游标我们必须提供一个起点–事件的ID我们将从该事件开始读取 如果未提供ID则光标将返回最早的可用事件。 因此每个读取器必须存储它已读取和处理的最后一个事件。 更重要的是如果没有新数据可用可尾光标可以有选择地阻塞一段时间从而解决了主动轮询问题。 顺便说一下mongo用于在副本集之间复制数据的oplog集合也是一个有上限的集合。从属Mongo实例在该集合后面尾随流式传输“事件”即数据库操作并按顺序在本地应用它们。 读取Java中的事件 使用Mongo Java驱动程序时 有一些“问题”。 首先您需要初始化游标。 为此我们需要提供1最后一个事件ID如果存在 2我们要读取事件的顺序此处为自然顺序即插入顺序 3两个关键的游标选项我们希望游标是可拖尾的并且如果没有新数据我们希望将其阻止 DBObject query lastReceivedEventId.isPresent()? BasicDBObjectBuilder.start(_id, BasicDBObjectBuilder.start($gte, lastReceivedEventId.get()).get()).get(): null;DBObject sortBy BasicDBObjectBuilder.start($natural, 1).get();DBCollection collection ... // must be a capped collection
DBCursor cursor collection.find(query).sort(sortBy).addOption(Bytes.QUERYOPTION_TAILABLE).addOption(Bytes.QUERYOPTION_AWAITDATA); 您可能想知道为什么我们使用 last_id而不是 。 由于生成Mongo ObjectId的方式在这里需要。 如果使用一个简单的 last_id我们可能会错过一些与last_id事件在同一秒之后但之后发生的事件。 这也意味着我们的Java代码必须处理这一事实并丢弃收到的第一个事件。 游标的类扩展了基本的Java Iterator接口因此非常易于使用。 因此现在我们可以进行批处理了。 在游标上进行迭代时驱动程序将批量从Mongo服务器接收数据 因此我们可以像调用其他迭代器一样简单地调用hasNext()和next()来接收后续元素并且只有某些调用会真正导致与服务器的网络通信。 在Mongo Java驱动程序中实际上可能阻塞的hasNext()是hasNext() 。 如果我们要分批处理事件我们需要a只要有可用的元素就读取它们并且b在被阻止没有更多事件之前有某种了解的方式并且我们可以处理事件已经批处理。 由于hasNext()可以阻止因此我们无法直接执行此操作。 这就是为什么我们引入了中间队列 LinkedBlockingQueue 的原因。 在单独的线程中从游标读取的事件在到达时即被放入队列中。 如果没有事件则线程将在cursor.hasNext()上cursor.hasNext() 。 阻塞队列有一个可选的大小限制因此如果队列已满则放置一个元素也将阻塞直到有可用空间为止。 在事件消费者线程中我们首先尝试以阻塞方式使用.poll从队列中读取单个元素因此我们在这里等待所有事件可用。 然后我们尝试将队列的全部内容消耗到一个临时集合中使用.drainTo 构建批处理并可能获取0个元素但我们始终拥有第一个。 值得一提的是如果集合为空则Mongo不会阻止因此我们必须回到主动轮询。 我们还必须考虑到游标可能会在等待期间死亡的事实。 要对此进行检查我们应该验证cursor.getCursorId() ! 0 其中0是“死光标”的ID。 在这种情况下我们只需要重新实例化游标即可。 加起来 综上所述我们得到了一个非常快速的事件源/流解决方案。 从某种意义上说这是“自我调节”即如果事件活动达到高峰则事件接收器将大批量读取这些事件。 如果事件活动少则将分批快速处理它们。 我们还将同一个Mongo实例用于其他目的。 从操作角度来看拥有一个数据库系统来聚簇和维护常规数据和事件肯定是一件好事。 参考 Adam Warski博客的Blog中来自我们的JCG合作伙伴 Adam Warski的MongoDB事件流 。 翻译自: https://www.javacodegeeks.com/2012/11/event-streaming-with-mongodb.html
