如何使用 RabbitMQ JMS 客户端从现有 JMS 应用程序连接到 RabbitMQ？

Question

我有一个通用的独立 JMS 应用程序，它与以下 JMS 提供程序 WebSphere、HornetQ 和 ActiveMq 一起使用。我将 Context.INITIAL_CONTEXT_FACTORY 和 Context.PROVIDER_URL 作为参数传递给我的应用程序，并通过执行类似的操作从它们中创建命名上下文

Properties environmentParameters = new Properties();
environmentParameters.put(Context.INITIAL_CONTEXT_FACTORY, property.context);
environmentParameters.put(Context.PROVIDER_URL, property.provider);
namingContext = new InitialContext(environmentParameters);

并使用此上下文进行对象查找。

我了解 RabbitMQ 不是 JMS 提供者，因此它没有 InitialContext 类或提供者 URL，但它提供了一个 JMS 客户端，它是符合 JMS 规范的 Java 客户端的抽象。 RabbitMQ 的 JMS 客户端 documentation 有一个将 JNDI 中的对象定义为作为 Web 应用程序的一部分的资源配置的示例，但我完全不知道如何为我的独立应用程序做类似的事情，该应用程序基于 JNDI 提供程序创建命名上下文使用 JMS 客户端的依赖项或从可用的依赖项中创建 InitialContext。

那么有人可以说明如何做到这一点吗？希望我的问题很清楚。

Answer 1

为了让 JMS 与 RabbitMQ 一起工作，您必须启用 插件 rabbitmq_jms_topic_exchange。
您可以按照本网站的说明下载（您需要登录）：
https://my.vmware.com/web/vmware/details?downloadGroup=VFRMQ_JMS_105&productId=349

1. 解压后，将文件 rjms-topic-selector-1.0.5.ez 放入文件夹 $RABBITMQ_SERVER\plugins。
2. 使用以下命令启用插件：rabbitmq-plugins enable rabbitmq_jms_topic_exchange
3. 使用以下命令检查插件是否运行正常：rabbitmq-plugins list
   
4. 重启 RabbitMQ - 我不确定是否真的有必要，但以防万一 ;-)
5. 在您的 RabbitMQ 网络管理 (http://localhost:15672/#/exchanges) 中，您可以查看可用的新 Exchange：
6. 现在，理论上 :-)，您已经能够使用标准 Java JMS API 连接到您的 RabbiMQ 服务器。

请记住，您必须创建一个 .bindings 文件才能让 JNDI 找到您注册的对象。这是它的内容示例：

ConnectionFactory/ClassName=com.rabbitmq.jms.admin.RMQConnectionFactory ConnectionFactory/FactoryName=com.rabbitmq.jms.admin.RMQObjectFactory ConnectionFactory/RefAddr/0/Content=jms/ConnectionFactory ConnectionFactory/RefAddr/0/Type=name ConnectionFactory/RefAddr/0/Encoding=String ConnectionFactory/RefAddr/1/Content=javax.jms.ConnectionFactory ConnectionFactory/RefAddr/1/Type=type ConnectionFactory/RefAddr/1/Encoding=String ConnectionFactory/RefAddr/2/Content=com.rabbitmq.jms.admin.RMQObjectFactory ConnectionFactory/RefAddr/2/Type=factory ConnectionFactory/RefAddr/2/Encoding=String # Change this line accordingly if the broker is not at localhost ConnectionFactory/RefAddr/3/Content=localhost ConnectionFactory/RefAddr/3/Type=host ConnectionFactory/RefAddr/3/Encoding=String # HELLO Queue HELLO/ClassName=com.rabbitmq.jms.admin.RMQDestination HELLO/FactoryName=com.rabbitmq.jms.admin.RMQObjectFactory HELLO/RefAddr/0/Content=jms/Queue HELLO/RefAddr/0/Type=name HELLO/RefAddr/0/Encoding=String HELLO/RefAddr/1/Content=javax.jms.Queue HELLO/RefAddr/1/Type=type HELLO/RefAddr/1/Encoding=String HELLO/RefAddr/2/Content=com.rabbitmq.jms.admin.RMQObjectFactory HELLO/RefAddr/2/Type=factory HELLO/RefAddr/2/Encoding=String HELLO/RefAddr/3/Content=HELLO HELLO/RefAddr/3/Type=destinationName HELLO/RefAddr/3/Encoding=String

然后……最后……代码：

属性 environmentParameters = new Properties(); environmentParameters.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.fscontext.RefFSContextFactory"); environmentParameters.put(Context.PROVIDER_URL, "file:/C:/rabbitmq-bindings"); 命名上下文 = 新的初始上下文（环境参数）； ConnectionFactory connFactory = (ConnectionFactory) ctx.lookup("ConnectionFactory");

Answer 2

对于遇到此异常的人

Caused by: javax.naming.NamingException: Unknown class [com.rabbitmq.jms.admin.RMQConnectionFactory]

即使在遵循@Ualter Jr. 的回答之后也是因为 .bindings 文件中的条目不正确。

我更正了 .bindings 文件中的以下 2 行

ConnectionFactory/ClassName=com.rabbitmq.jms.admin.RMQConnectionFactory --->
ConnectionFactory/ClassName=javax.jms.ConnectionFactory

和

YourQueueName/ClassName=com.rabbitmq.jms.admin.RMQDestination --->
StriimQueue/ClassName=javax.jms.Queue

当我再次遇到这个异常时，我刚刚打开了这个class，发现它需要以下类名

      /*
     * Valid class names are:
     * javax.jms.ConnectionFactory
     * javax.jms.QueueConnectionFactory
     * javax.jms.TopicConnectionFactory
     * javax.jms.Topic
     * javax.jms.Queue
     *
     */

更正这些条目将使现有/新的 JMS 应用程序能够与 RabbitMQ 一起使用。

Answer 3

我们可以使用下面的 java 代码为 RabbitMQ 生成 .bindings 文件：

import java.util.Properties;
import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.Reference;
import javax.naming.StringRefAddr;

Properties env = new Properties();
    env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.fscontext.RefFSContextFactory");
    env.put(Context.PROVIDER_URL, "file:bindings");
    Context ctx = new InitialContext(env);

    Reference connectionFactoryRef = new Reference(ConnectionFactory.class.getName(), RMQObjectFactory.class.getName(), null);
    connectionFactoryRef.add(new StringRefAddr("name", "jms/ConnectionFactory"));
    connectionFactoryRef.add(new StringRefAddr("type", ConnectionFactory.class.getName()));
    connectionFactoryRef.add(new StringRefAddr("factory", RMQObjectFactory.class.getName()));
    connectionFactoryRef.add(new StringRefAddr("host", "$JMS_RABBITMQ_HOST$"));
    connectionFactoryRef.add(new StringRefAddr("port", "$JMS_RABBITMQ_PORT$"));
    ctx.rebind("ConnectionFactory", connectionFactoryRef);

    String jndiAppend = "jndi";
    for (int i = 1; i <= 10; i++) {
        String name = String.format("queue%02d", i);
        Reference ref = new Reference(Queue.class.getName(), com.rabbitmq.jms.admin.RMQObjectFactory.class.getName(), null);
        ref.add(new StringRefAddr("name", "jms/Queue"));
        ref.add(new StringRefAddr("type", Queue.class.getName()));
        ref.add(new StringRefAddr("factory", RMQObjectFactory.class.getName()));
        ref.add(new StringRefAddr("destinationName", name));
        ctx.rebind(name+jndiAppend, ref);

        name = String.format("topic%02d", i);
        ref = new Reference(Topic.class.getName(), com.rabbitmq.jms.admin.RMQObjectFactory.class.getName(), null);
        ref.add(new StringRefAddr("name", "jms/Topic"));
        ref.add(new StringRefAddr("type", Topic.class.getName()));
        ref.add(new StringRefAddr("factory", RMQObjectFactory.class.getName()));
        ref.add(new StringRefAddr("destinationName", name));
        ctx.rebind(name+jndiAppend, ref);
    }