Java，使用队列管理器处理海量消息（交易）答案

【问题标题】：Java, Massive message processing with queue manager (trading)Java，使用队列管理器处理海量消息（交易）
【发布时间】：2010-12-19 12:06:46
【问题描述】：

我想设计一个简单的应用程序（没有 j2ee 和 jms），可以处理大量的消息（比如在交易系统中）

我创建了一个服务，可以接收消息并将它们放入队列中，以便系统在过载时不会卡住。

然后我创建了一个包装队列的服务（QueueService），并有一个从队列中弹出消息的pop方法，如果没有消息返回null，则将此方法标记为“已同步”以进行下一步。

我创建了一个知道如何处理消息的类 (MessageHandler) 和另一个可以在新线程中“侦听”消息的类 (MessageListener)。该线程有一个“while(true)”，并且一直试图弹出一条消息。

如果返回了一条消息，线程调用 MessageHandler 类，完成后，他会请求另一条消息。

现在，我已将应用程序配置为打开 10 个 MessageListener 以允许多消息处理。

我现在有 10 个线程一直处于循环状态。

这是一个好的设计吗？？

任何人都可以参考一些书籍或网站如何处理这种情况吗？

谢谢，罗尼

【问题讨论】：

谢谢大家，这一切都非常有帮助。我真的要研究 JMS 并探索他的选择，但现在我只想为我们正在构建的一个小型应用程序创建一个 POC。再次感谢罗尼
下载一个开源 JMS 并使用它。有很多事情需要考虑，你无法在合理的时间内独自完成。您不需要 Container 来运行 JMS。如果你想要事务，你确实需要某种事务管理器（这就是容器派上用场的地方）

标签： java multithreading architecture trading

【解决方案1】：

虽然它现在已经过时了，但Practical .NET for Financial Markets 展示了您在开发金融交易系统时应该考虑的一些通用概念。虽然它面向 .Net，但您应该能够将一般概念转换为 Java。

【讨论】：

【解决方案2】：

从字里行间看，这听起来像是您没有使用诸如 MQ 之类的 JMS 提供程序。您的解决方案在大多数情况下听起来都不错，但是我会质疑您不使用 JMS 的原因。

你提到了一些关于交易的事情，我可以确认很多交易系统都使用 JMS 和不带 j2ee。如果您真的想要高性能、可靠性和头脑清醒，请不要通过编写自己的排队系统来重新发明轮子，请查看一些 JMS 提供程序及其客户端 API。

卡尔

【讨论】：

+1 - 重新发明 API 只是为了满足自己只会在以后带来更多困难。

【解决方案3】：

我知道像 Websphere 和 Sonic 这样的队列服务提供商需要花钱，但总是有 JBoss Messaging、FUSE 和 ApacheMQ 等。不要尝试制作比 JMS 更好的 JMS。大多数 JMS 提供程序都具有持久性功能，可在队列或应用程序服务器死机时提供容错功能。不要重新发明轮子。

【讨论】：

【解决方案4】：

监听消息和处理消息的分离对我来说似乎是明智的。拥有可扩展数量的处理线程也很好，您可以根据自己的平台上的并行处理工作量来调整数量。

我不太满意的一点是线程轮询消息到达的方式 - 在这里您正在做繁忙的工作，如果您添加睡眠以减少这种情况，那么您不会立即对消息到达做出反应。 JMS API 和 MDB 采用更多的事件驱动方法。我将看看它是如何在开源 JMS 中实现的，以便您可以看到替代方案。 [我也赞同为自己重新发明 JMS 可能是个坏主意的观点。]要记住的是，随着您的系统变得越来越复杂，您会添加更多队列并且处理更多繁忙的工作会产生更大的影响。

我担心的另一个问题是你会遇到使用单台机器的限制，首先你可以允许更大的可伸缩性，允许监听器在多台机器上。其次，您有单点故障。显然，解决这类问题是消息传递供应商赚钱的地方。这也是为什么购买而不是构建更容易成为复杂中间件的另一个原因。

【讨论】：

【解决方案5】：

从您的描述看来，您走在正确的道路上，只有一点点例外。您在从队列中检索消息时实现了忙碌等待。

更好的方法是在同步的popMessage() 方法中阻塞线程，在无法弹出更多消息时对队列资源执行wait()。当向队列中添加（a）条消息时，等待的线程通过notifyAll()被唤醒，一个或多个线程将收到一条消息，其余线程重新进入wait()状态。

这样CPU资源的分配会更顺畅。

【讨论】：

【解决方案6】：

您需要非常轻便、超快速、可扩展的排队系统。试试Hazelcast分布式队列！

它是java.util.concurrent.BlockingQueue 的分布式实现。详情请查看documentation。

Hazelcast 实际上不仅仅是一个分布式队列；它是 Java 的队列、主题、映射、多映射、锁、执行器服务的事务性分布式实现。

它是在 Apache 许可下发布的。

【讨论】：

【解决方案7】：

事件循环

改用event loop/message pump 怎么样？实际上，我是通过观看 Ryan 精彩的 node.js 视频演示来学习这项技术的，我认为如果还没有观看的话，您应该真的观看。

您最多将 10 条消息从 Thread a 推送到 Thread b（如果已满则阻塞）。线程 a 有一个无限的[LinkedBlockingQueue][3]()。线程 b 有一个大小为 10 (new ArrayBlockingQueue(10)) 的有界 [ArrayBlocking][4]。 thread a 和 thread b 都有一个无限的“while 循环”。 Thread b 将处理来自ArrayBlockingQueue 的消息。这样，您将只有 2 个无休止的“while 循环”。作为旁注，在阅读规范时使用 2 个 arrayBlockingQueues 可能会更好，因为以下句子：

链接队列通常具有更高的吞吐量高于基于数组的队列，但大多数情况下的可预测性较差并发应用程序。

当然，数组支持的队列有一个缺点，它会使用更多内存，因为您必须事先设置大小（太小不好，因为它会在满时阻塞，太大也可能是一个问题，如果低内存）使用。

接受的解决方案：

在我看来，您应该更喜欢我的解决方案而不是公认的解决方案。原因是如果一切可能，您应该只使用java.util.concurrent 包。编写正确的线程代码很难。当你犯错时，你最终会陷入僵局、饥饿等。

Redis：

就像其他人已经提到的那样，您应该为此使用 JMS。我的建议与此类似，但我认为使用/安装更简单。首先，我假设您的服务器正在运行 Linux。我建议你安装 Redis。 Redis 真的很棒/很快，你也应该把它用作你的数据存储。它具有您可以使用的阻止列表操作。 Redis 会将您的结果存储到磁盘中，但以一种非常有效的方式。

祝你好运！

【讨论】：