草庐IT

mongodb - 什么是 MongoDB 更改流读取器的良好水平扩展策略？

coder 2023-10-31 原文

我正在考虑实现一个 MongoDB 变更流读取器，我想确保我做的是正确的。有很多关于如何实现实际阅读器代码的简单示例，包括 official documentation ，我不太担心这方面的问题。

不过，我有点担心读者落后于变化流，无法 catch 变化流，我想确保读者能够处理变化流。

mongo 服务器是一个集群，为了便于论证，我们假设它在一天中的所有时间都非常繁忙。更改流 API 似乎只与执行工作的单个实例兼容，因为它必须如何迭代流结果而不是像队列一样对其进行操作。因此，我担心迭代结果的单个实例可能比将新项目推送到流中花费更长的时间来完成它的工作。

我的直觉是让读者简单地读取流，批处理更改，然后将其推送到队列中，其他工作人员可以在队列中横向扩展以完成工作。但是，我仍然有一个作为读取器的实例，理论上它仍然有可能落后于流，即使只做了将更改推送到队列中的最低限度的工作。

所以我的问题是，这种担心有多现实？是否有任何方法可以创建读取器，即使它只是将更改流式传输到工作队列中，它也可以水平扩展？我还应该考虑哪些其他注意事项？

最佳答案

最有可能的是，只需将所有工作委派给水平缩放的队列，单个阅读器就足够了。

如果事实证明这还不够，并且您的读者仍然需要水平缩放，那么您可以通过使用匹配过滤器来实现这一点，这种方式允许多个读者分担工作。

例如，如果您有一个包含十六进制字符的 ID，您可以通过在每个服务器上使用匹配运算符将工作拆分到两个服务器上，其中每个服务器匹配整个范围内的一半字符:

// Change Stream Reader 1
const params = [
  { $match: { _id: /^[0-7]/ } }
];
const collection = db.collection('inventory');
const changeStream = collection.watch(params);

在第二台机器上:

// Change Stream Reader 2
const params = [
  { $match: { _id: /^[8-9a-f]/ } }
];
const collection = db.collection('inventory');
const changeStream = collection.watch(params);

如果您需要超过 16 台服务器，那么您可以使范围更加具体:

// Server 0  matches on /^0[0-7]/
// Server 1  matches on /^1/
// ...
// Server 15 matches on /^f/
// Server 16 matches on /^0[8-9a-f]/

这将允许每台机器观看消息的子集并处理它们，而其他机器正在处理其他消息而不会重复。

以稳健的方式协调哪个服务器正在监视哪个范围变得有些复杂，因为您需要确保崩溃或挂起的机器恢复，如果您需要动态地水平扩展，那么您需要能够向服务器提供新的范围并调整大小。此解决方案还会导致消息被乱序处理，因此如果顺序很重要，那么您需要想出一个解决方案来重新排序消息或处理乱序问题。

但是这些都是与这个问题不同的主题，所以我现在将省略细节。

关于mongodb - 什么是 MongoDB 更改流读取器的良好水平扩展策略？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/54295043/

读取器良好sectioncollectionmongodbchangestream

有关mongodb - 什么是 MongoDB 更改流读取器的良好水平扩展策略？的更多相关文章

  1. ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2

    类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc

  2. ruby-on-rails - Ruby on Rails 迁移，将表更改为 MyISAM - 2

    如何正确创建Rails迁移，以便将表更改为MySQL中的MyISAM？目前是InnoDB。运行原始执行语句会更改表，但它不会更新db/schema.rb，因此当在测试环境中重新创建表时，它会返回到InnoDB并且我的全文搜索失败。我如何着手更改/添加迁移，以便将现有表修改为MyISAM并更新schema.rb，以便我的数据库和相应的测试数据库得到相应更新？ 最佳答案 我没有找到执行此操作的好方法。您可以像有人建议的那样更改您的schema.rb，然后运行:rakedb:schema:load，但是，这将覆盖您的数据。我的做法是(假设

  3. ruby-on-rails - Rails - 子类化模型的设计模式是什么？ - 2

    我有一个模型:classItem项目有一个属性“商店”基于存储的值，我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式？如果方法中没有大的if-else语句，这是如何干净利落地完成的？ 最佳答案 通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.co

  4. ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2

    我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚

  5. ruby - 什么是填充的 Base64 编码字符串以及如何在 ruby​​ 中生成它们？ - 2

    我正在使用的第三方API的文档状态:"[O]urAPIonlyacceptspaddedBase64encodedstrings."什么是“填充的Base64编码字符串”以及如何在Ruby中生成它们。下面的代码是我第一次尝试创建转换为Base64的JSON格式数据。xa=Base64.encode64(a.to_json) 最佳答案 他们说的padding其实就是Base64本身的一部分。它是末尾的“=”和“==”。Base64将3个字节的数据包编码为4个编码字符。所以如果你的输入数据有长度n和n%3=1=>"=="末尾用于填充n%

  6. ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2

    我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i

  7. ruby - 为什么 4.1%2 使用 Ruby 返回 0.0999999999999996？但是 4.2%2==0.2 - 2

    为什么4.1%2返回0.0999999999999996？但是4.2%2==0.2。 最佳答案 参见此处:WhatEveryProgrammerShouldKnowAboutFloating-PointArithmetic实数是无限的。计算机使用的位数有限(今天是32位、64位)。因此计算机进行的浮点运算不能代表所有的实数。0.1是这些数字之一。请注意，这不是与Ruby相关的问题，而是与所有编程语言相关的问题，因为它来自计算机表示实数的方式。 关于ruby-为什么4.1%2使用Ruby返

  8. ruby - 使用 C 扩展开发 ruby​​gem 时，如何使用 Rspec 在本地进行测试？ - 2

    我正在编写一个包含C扩展的gem。通常当我写一个gem时，我会遵循TDD的过程，我会写一个失败的规范，然后处理代码直到它通过，等等......在“ext/mygem/mygem.c”中我的C扩展和在gemspec的“扩展”中配置的有效extconf.rb，如何运行我的规范并仍然加载我的C扩展？当我更改C代码时，我需要采取哪些步骤来重新编译代码？这可能是个愚蠢的问题，但是从我的gem的开发源代码树中输入“bundleinstall”不会构建任何native扩展。当我手动运行rubyext/mygem/extconf.rb时，我确实得到了一个Makefile(在整个项目的根目录中)，然后当

  9. Ruby 写入和读取对象到文件 - 2

    好的，所以我的目标是轻松地将一些数据保存到磁盘以备后用。您如何简单地写入然后读取一个对象？所以如果我有一个简单的类classCattr_accessor:a,:bdefinitialize(a,b)@a,@b=a,bendend所以如果我从中非常快地制作一个objobj=C.new("foo","bar")#justgaveitsomerandomvalues然后我可以把它变成一个kindaidstring=obj.to_s#whichreturns""我终于可以将此字符串打印到文件或其他内容中。我的问题是，我该如何再次将这个id变回一个对象？我知道我可以自己挑选信息并制作一个接受该信

  10. ruby - ruby 中的 TOPLEVEL_BINDING 是什么？ - 2

    它不等于主线程的binding，这个toplevel作用域是什么？此作用域与主线程中的binding有何不同？>ruby-e'putsTOPLEVEL_BINDING===binding'false 最佳答案 事实是，TOPLEVEL_BINDING始终引用Binding的预定义全局实例，而Kernel#binding创建的新实例>Binding每次封装当前执行上下文。在顶层，它们都包含相同的绑定(bind)，但它们不是同一个对象，您无法使用==或===测试它们的绑定(bind)相等性。putsTOPLEVEL_BINDINGput

随机推荐