消息驱动的 bean (MDB) 可以在“外部”MQ 上侦听吗？

Posted 2023-03-10

【中文标题】消息驱动的 bean (MDB) 可以在“外部”MQ 上侦听吗？

【英文标题】：Can Message driven beans (MDB) listen on "external" MQ?

【发布时间】：2015-09-20 10:42:35

【问题描述】：

我正在尝试理解与 MDB、MQ、JMS 相关的概念。在问这个问题之前，我对 SO 进行了研究。

这是可能的情况吗：

MDB 部署在 Application Server 上，比如 JBOSS（在物理 Server-A 上）。

MQ（比如 ApacheMQ）在不同的物理服务器 B 上。

那么部署在物理服务器-A的MDB能否从物理服务器-B获取消息？

如果可能，那么 MDB 是否使用 JMS API？

我听说 Jboss 有 MQ，我假设 MQ 带有 Jboss 应用服务器；但是我希望 MDB 在不同的服务器上，MQ 服务器在不同的物理服务器上。

感谢您帮助理解这一点。

【参考方案1】：

这有 4 个真实的部分。

1) 您的应用程序。 MDB 是一个实现 J2EE 消息驱动 Bean API 的应用程序，最简单的形式意味着它有一个 onMessage() 函数，当消息到达时将由应用程序服务器调用。

2) J2EE应用服务器，JBOSS就是一个例子。它从 MQ 客户端接收消息，并将它们转发到 MDB。

3) MQ 客户端。这是由实现 J2EE 的 JCA RA 和 JMS 部分的 MQ 提供程序（IBM/Apache/etc）编写的代码。应用程序可以通过 JMS 与此客户端交互以放置和获取消息，但当您作为 MDB 进行交互时，您将通过 onMessage() 方法进行驱动。此客户端将消息传递给驱动应用程序的应用程序服务器。

4) MQ 服务器。 IBM MQ 将此称为“队列管理器”，它可以存在于任何地方。 #3 的客户端将通过网络连接到队列管理器。

#1、#2 和 #3 需要在同一台物理机器上（并在同一 JVM 中运行）。 #4 可以在任何地方通过网络访问。

解决您的问题：

那么部署在物理服务器-A的MDB能否从物理服务器-B获取消息？

是的

如果可能，那么 MDB 是否使用 JMS API？

MDB 由应用服务器使用 J2EE API 驱动，JMS 只是其中的一部分。

顺便说一句，“MQ”是产品的名称，而不是概念。通用名称是“消息提供者”。 IBM MQ 和 ApacheMQ 都是消息传递提供者。

【讨论】：

很好的解释。谢谢。

【参考方案2】：

由远程 MQ 实例激活的 MDB 示例：

/**
 * WebSphereMQ.java
 * 
 * Created on Sep 21, 2012, 9:11:29 AM
 *
 * To the extent possible under law, Red Hat, Inc. has dedicated all copyright to this 
 * software to the public domain worldwide, pursuant to the CC0 Public Domain Dedication. This 
 * software is distributed without any warranty.  
 *
 * See <http://creativecommons.org/publicdomain/zero/1.0/>.
 *
 */
package org.jboss.sample.mq;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.CreateException;
import javax.ejb.EJBException;
import javax.ejb.MessageDriven;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.jboss.ejb3.annotation.Pool;
import org.jboss.ejb3.annotation.ResourceAdapter;
import org.jboss.logging.Logger;

/**
 * 
 */
@MessageDriven(name = "WebSphereMQ", activationConfig = 
        @ActivationConfigProperty(propertyName = "maxPoolDepth", propertyValue="100"),
        @ActivationConfigProperty(propertyName = "maxMessages", propertyValue="1"),
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "hostName", propertyValue = "10.0.0.150"),
        @ActivationConfigProperty(propertyName = "port", propertyValue = "1414"),
        @ActivationConfigProperty(propertyName = "userName", propertyValue = "redhat"),
        @ActivationConfigProperty(propertyName = "password", propertyValue = "redhat"),
        @ActivationConfigProperty(propertyName = "channel", propertyValue = "SYSTEM.DEF.SVRCONN"),
        @ActivationConfigProperty(propertyName = "queueManager", propertyValue = "REDHAT.QUEUE.MANAGER"),
        @ActivationConfigProperty(propertyName = "useJNDI", propertyValue = "true"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "jms/queue/gssQueue"),
        @ActivationConfigProperty(propertyName = "transportType", propertyValue = "CLIENT") )

@Pool(value="MQpool") 
@ResourceAdapter(value="wmq.jmsra.7.5.0.4.rar")
@TransactionAttribute(TransactionAttributeType.REQUIRED)
public class WebSphereMQ implements MessageListener 
    private static final Logger logger = Logger.getLogger(WebSphereMQ.class);

    /*
     * (non-Javadoc)
     * 
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
     */
    public void onMessage(Message message) 
        try 
            logger.info("Received message: " + message.getJMSMessageID() + " : " + ((TextMessage)message).getText());

            try            
                Thread.sleep(10000);
             catch(Exception e) 
                logger.info("interrupted");
                           
    

【参考方案3】：

对于使用 Glassfish4/Payara-Server 的人，这相当于 @Doug Grove @ResourceAdapter ：

@MessageDriven(
  name = "foo",
  activationConfig = 
     @ActivationConfigProperty(propertyName = "resourceAdapter", propertyValue = "activemq-rar-x-x-x"),
:
:

当activemq-rar-x-x-x是部署在AS的ActiveMQ资源适配器。

问候。