JMS 消息驱动 Bean 工作同步

Question

我们刚刚开始构建 JMS 架构并具有以下基本设置：

1. GLassfish v2.1
2. MDB 通过 TopicConnectionFactory 监听主题（全部在本地服务器上）

现在，当新消息到达时，MDB 会生成一个工作线程，即使我们要按顺序传递消息，我们需要一个同步机制，以便线程在并发处理请求之前检查特定条件。

有没有办法让这些线程共享数据？ 或者是否有任何其他机制（除了数据库表/行锁）可以用于同步？

提前致谢。

---

澄清一下，我不是在创建自己的线程。 正如每个人正确指出的那样，容器为我做到了这一点。 让我用一个例子来帮助解释我的困境。

-消息 A 在 t=0 时到达，“创建”数据 id 1 -

消息 B 在 t=0.1 时到达，“更新”数据 id 1

现在假设容器生成 2 个工作进程来处理 A 和 A。 B 并且“创建”数据比更新数据花费更多的时间，更新会更早处理并且没有效果。

更清楚地说，

-在处理消息 B 时，我会在 t=1 处查找数据 id 1（未找到它，因此无需执行任何操作即可完成）。

-数据 ID 1 将在 t=2 处处理消息 A 时创建。

Answer 1

学究警报！ 我是那种会阅读技术实际规范的人。

阅读 EJB 规范 版本 3.0，第 21.1.2 节（编程限制）不允许在您的代码。 这是语言和理由......

企业 Bean 不得尝试
来管理线程。 企业bean
不得尝试启动、停止、
挂起或恢复线程，或
更改线程的优先级或名称。
企业 Bean 不得尝试
管理线程组。

这些函数是为 EJB 容器保留的。 让企业
管理线程的 bean 会减少
容器能够正确地
管理运行时环境。

因此，如果您按照您所说的去做，EJB 警察就会在半夜来敲您的门并把您带走。 或者你的应用程序可能会出现故障，当你抱怨时供应商会笑。 或者根本不会发生任何不好的事情。

但是，正如达菲莫所说，为什么要这样做呢？ 如果您想要大量线程提供的可伸缩性，您可以为您的 MDB 进行配置吗？ EJB 的目的是为您处理类似的事情。

Answer 2

我不明白为什么 MDB 必须产生一个工作线程。 JMS 中有一个与消息侦听器关联的线程池。 这就是应该完成这项工作的线程。

EJB 规范规定 Bean 中不会产生线程。 容器处理线程。 这也包括多边开发银行。

侦听器应该处理从队列中取出的消息。 它需要的数据应该在消息中。 有什么必要分享？

我认为您的方法违背了推荐的 EJB 实践。

Answer 3

真正的问题是，应用程序服务器通常会自行生成上述工作线程。
虽然 JMS 保证消息的消费顺序至少与在一个生产者中生成消息的顺序相同，但 MDB 规范明确指出不会保留该顺序（因为提到了工作线程）。 请参阅 JSR-000220 Enterprise JavaBeans 3.0 最终版本 (ejbcore)。

没有可移植且 100% 可靠的方法来规避此问题。 由于工人的性质，引入了无法控制的竞争条件。

幸运的是，大多数应用程序服务器都有配置工作线程数量的方法，尽管是专有的且不兼容。

对于 JBoss 和 ActiveMQ，这是可行的：

@PoolClass(value = org.jboss.ejb3.StrictMaxPool.class, maxSize = 1)
@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
    @ActivationConfigProperty(propertyName = "maxSessions", propertyValue = "1"),
    @ActivationConfigProperty(propertyName = "destination", propertyValue = "TEST.FOO")
})
@ResourceAdapter("activemq-ra.rar")
public class NewMessageBean implements MessageListener { ... }

在这种情况下，“maxSessions”是工作人员的数量。 它可能与其他 JMS 提供商不同，但这应该指向正确的方向。

Answer 4

如上所述，JMS 框架处理调度问题，例如分派线程。 您在此所做的任何操作不仅不如默认行为，而且可能会严重限制 JMS 的功能。

更复杂的 JMS 处理程序被设计为跨多个节点（= 服务器）工作，因此任何共享内存解决方案都会将您限制为单个节点上的单个 JVM，这将是一个遗憾，因为 JMS 的巨大优势是可扩展性。

一个可能的 JMSy 解决方案是使用一个“cookie”队列，将单个虚拟“cookie”消息用于同步活动。
当您的进程执行有争议的活动时，它会“等待”来自“cookie”队列的单个消息，当有争议的工作完成时，它会将 cookie 放回到队列中。 JMS 的魔力将处理几乎所有的阻塞、等待和错误恢复。

Answer 5

看来在每个人都回答之后，问题的性质完全改变了。 我将为后代做出一个迟来的回应； 这假设有某种类型的消息 ID 可用于排序。 您说“我们已按顺序传递消息”，但您没有确切说明这是如何实现的。

...由于您已经澄清您没有启动自己的线程，因此基本上在数据@ ID=1 的“创建”与其后续的“更新”之间存在竞争条件。 我假设您在创建和/或更新数据时锁定数据 @ ID=1 。 因此，有两种可能性：

1. “创建数据”消息首先到达：(1) 锁 ID=1，(2) 创建数据。 (3)释放锁。 (4) 应用待处理的“更新”。
2. “更新”在“创建”之前到达：(1) 锁 ID=1，(2) 创建丢失的数据（即，执行“更新插入”：插入数据，即使数据不存在）。 (3)释放锁。 (4) 忽略挂起的“create data”消息（“create”消息的序列号比“update”消息的序列号低）
   • 如果消息发送者被允许同时发送更新和插入（没有请求/响应），那么它们确实应该有某种序列号。 如果是这样，当“插入”到达时，其序列号将小于该行中的当前值，因此可以忽略它。 或者，如果“创建”消息类型与“更新”消息类型不同，那么如果数据已存在，则始终可以忽略“创建”。

我认为您遇到的问题是如何同步数据。 基本上，线程共享对象并创建互斥体（互斥）以允许单个线程访问数据，从而导致另一个线程阻塞。 这可以简单地通过 Java 的低级同步工具（“synchronized”关键字）或辅助此操作的内置类来完成（http://java.sun.com/docs/books/tutorial/essential/concurrency/highlevel.html）。

Answer 6

正如一些人提到的，您不应该在 MDB（或任何其他类型的 EJB 或与此相关的 servlet）中创建自己的线程。

许多 EJB 容器实际上不允许您创建和运行线程。 不过，有一种安全的方法可以做到这一点，即使用 commonj 规范中的 WorkManager，尽管在这种特殊情况下我看不出有什么理由这样做，因为 MDB 已经在它自己的“工作线程”中运行。

请参阅有关生成线程的信息了解更多信息有关为什么不应在 Java EE 服务器中生成线程以及如何在需要时安全地执行此操作的信息。