在Rails中执行大量的异步IO绑定操作

Question

我正在研究定期需要执行大量IO绑定操作的Rails应用程序。这些操作可以异步执行。例如，每天一次，对于每个用户，系统需要查询Salesforce.com以获取用户当前正在跟踪的帐户（公司）列表。这导致大量的数据（可能大于10万）的小型查询。

我们目前的方法是使用ActiveMQ和ActiveMessaging。我们的每个用户都会作为不同的消息被推送到队列中。然后，消费者将用户从队列中拉出，查询Salesforce.com并处理结果。但是这种方法给我们带来了可怕的表现。在单个轮询器流程中，我们一次只能处理一个用户。因此，Salesforce.com查询将被序列化。除非我们直接运行数百个轮询器进程，否则我们无法接近饱和运行轮询器的服务器。

我们正在寻找EventMachine作为替代方案。它的优点是允许我们在单个EventMachine进程中同时启动大量的Salesforce.com查询。所以，我们得到了很好的并行性和我们的服务器的利用率。

但EventMachine存在两个问题。 1）我们使用ActiveMQ/ActiveMessaging失去了可靠的消息传递。 2）我们无法定期轻松重启我们的EventMachine以减少内存增长的影响。例如，使用ActiveMessaging，我们有一个cron作业，每天重新启动一次轮询器，这可以在不担心丢失任何消息的情况下完成。但是通过EventMachine，如果我们重新启动流程，我们可能会丢失数百封正在进行的消息。我能看到的唯一方法是在EventMachine之上构建一个持久/可靠的传递层。

有没有人有更好的方法？可靠地执行大量异步IO绑定操作的最佳方式是什么？

Answer 1

我维护ActiveMessaging，并且一直在考虑多线程轮询器的问题，尽管这些问题可能与您所处的规模不同。我会在这里给你我的想法，但也很乐意进一步讨论活跃的消息列表，或者如果你喜欢，可以通过电子邮件进行讨论。

一个窍门是轮询器不是这个的唯一序列化部分。 STOMP订阅，如果您执行客户端 - > ack以防止在中断时丢失消息，则只有在先前的消息已被删除时才会在给定的连接上发送新消息。基本上，每个连接一次只能处理一条消息。

所以要继续使用代理，诀窍就是让许多代理连接/订阅一次打开。当前的轮询器对此非常沉重，因为它加载了每个轮询器的整个轨道环境，并且一个轮询器是一个连接。但是对于当前的轮询者来说，没有什么不可思议的，我可以想象将一个轮询器写成一个事件机器客户端，它被实现为创建与代理的新连接并同时获得许多消息。在我最近的实验中，我一直在考虑使用Ruby Enterprise Edition，并且有一个主线程可以派生许多轮询工作者线程，以获得减少内存占用的好处（就像乘客一样），但是我认为EM技巧也可以起作用。

我也是Resque项目的崇拜者，虽然我不知道它会更好地扩展到许多工人 - 我认为工人可能更轻。

http://github.com/defunkt/resque

Answer 2

我使用AMQP与RabbitMQ的的方式，会为你工作。由于ActiveMQ实现了AMQP，我想你可以以类似的方式使用它。我没有使用ActiveMessaging，虽然它看起来像一个很棒的包，但我怀疑可能不适合这个用例。

这里是你如何能做到这一点，使用AMQP：

• 有无Rails的进程发送一个消息，说“得到用户i信息”。
• 消费者从消息队列中取消此消息，确保指定该消息需要将'ack'从队列中永久删除。这意味着如果消息没有被确认为已处理，它最终会返回到另一个工作人员的队列中。
• 工作人员然后将消息转入SalesForce的数千个小请求中。
• 当所有这些请求都成功返回时，应该触发另一个回调来确认原始消息并返回一个“摘要消息”，该消息包含与原始请求密切相关的所有信息。关键是使用一个消息队列，让您确认成功处理给定的消息，并确保只在相关处理完成时才这样做。
• 另一名工作人员将该消息从队列中拉出并执行适当的同步工作。由于所有的延迟引起的位已经执行，我想这应该没问题。

如果您使用（C）Ruby，请尽量不要在同一个进程中同步和异步。一个进程应该通过Eventmachine来完成所有的事情，不需要阻塞代码，或者只通过消息队列与Eventmachine进程通信。另外，编写异步代码非常有用，但也很难编写，难以测试，并且容易出错。小心。如果适用，请使用其他语言或工具进行调查。

Answer 3

也会检出“抽筋”和“魔豆”

Answer 4

有人给我发以下链接：http://github.com/mperham/evented/tree/master/qanat/。这是一个有点类似于ActiveMessaging的系统，只是它建立在EventMachine之上。这几乎就是我们所需要的。唯一的问题是，它似乎只适用于Amazon的队列，而不是ActiveMQ。