javajms消息驱动的Bean工作者同步

Question

我们刚刚开始构建JMS体系结构，并有以下基本设置：

1. 玻璃鱼v2。一,
2. MDB通过TopicConnectionFactory监听主题（全部在本地服务器上）

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

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

提前谢谢

---

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

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

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

现在假设容器产生2名工人来处理A&；B而且“创建”数据比更新数据要花更多的时间，更新过程会更早，而且没有效果

更清楚地说

-在处理消息B时，我会在t=1处查找数据id 1（没有找到它，因此不做任何事情就完成了）

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

Answer 1

学究警觉！我是那种阅读技术规范的人

阅读EJB spec版本3.0，第21.1.2节（编程限制）不允许在代码中使用线程。以下是语言和基本原理

The enterprise bean must not attempt to manage threads. The enterprise bean must not attempt to start, stop, suspend, or resume a thread, or to change a thread’s priority or name. The enterprise bean must not attempt to manage thread groups.

These functions are reserved for the EJB container. Allowing the enterprise bean to manage threads would decrease the container’s ability to properly manage the runtime environment.

所以如果你照你说的去做，EJB警察会在半夜敲门，把你带走。或者你的应用程序可能会出现故障，当你抱怨时，供应商会大笑。否则什么坏事都不会发生

但是，正如达菲莫所说，为什么要这样做？如果你想要大量线程提供的可伸缩性，你能为你的MDB配置吗？EJB的重点是为您处理这样的事情

Answer 2

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

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

有关为什么不应该在JavaEE服务器中生成线程以及在需要时如何安全地生成线程的更多信息，请参见info on spawning threads here

Answer 3

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

EJB规范说bean中没有线程生成。容器处理线程。这也包括MDB

侦听器应该正在处理从队列中删除的消息。它需要的数据应该在消息中。分享的必要性是什么

我认为您的方法与推荐的EJB实践背道而驰

Answer 4

如上所述，JMS框架处理调度问题，例如调度线程。您在这方面所做的任何事情都不仅不如默认的beheaviour，而且可能会严重限制JMS的功能

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

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

Answer 5

在每个人都回答了这个问题之后，这个问题的性质似乎完全改变了。我将为子孙后代做出一个姗姗来迟的回应；这假设存在某种类型的消息ID可用于订购。你说“我们按顺序传递信息”，但你没有具体说明这是如何实现的

。。。既然你已经澄清了你不是在启动你自己的线程，你基本上在数据@ID=1的“创建”和随后的“更新”之间有一个竞争条件。我假设您在创建和/或更新数据@ID=1时锁定了该数据。因此，有两种可能性：

1. “创建数据”消息首先到达：（1）lock ID=1，（2）创建数据。（3） 松开锁。（4） 应用挂起的“更新”
2. “update”在“create”之前到达：（1）lock ID=1，（2）创建缺少的数据（即，执行“upsert”：即使数据不在那里也插入数据）。（3） 松开锁。（4） 忽略挂起的“创建数据”消息（“创建”消息的序列号低于“更新”
   • 如果允许消息发送者同时发送更新和插入（无需请求/响应），那么它们确实应该有某种序列号。如果是这样，当“insert”到达时，其序列号将小于该行中的当前值，因此可以忽略它。或者，如果“创建”消息类型不同于“更新”消息类型，那么如果数据已经存在，则“创建”总是可以忽略

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

Answer 6

真正的问题是，应用程序服务器通常会自行产生上述工作人员。 虽然JMS保证消息的使用顺序与它们至少在一个生产者中产生的顺序相同，但MDB规范明确指出，该顺序不会被保留（因为提到的工人）。见JSR-000220 Enterprise JavaBeans 3.0 Final Release (ejbcore)第5.4.11节

没有一种便携且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提供商不同，但这应该指向正确的方向