在AMQP组件的Apache Camel XML DSL中配置Apache Qpid JMS（jakarta.jms）连接工厂

Question

尝试使用jakarta.jmsApache Qpid AMQP客户端来处理消息。

我正在尝试使用org.messagehub使用可池连接工厂。

独立java代码工作，请参阅java更改。当我尝试在Camel中使用相同的Spring-XML-DSL时，AMQP组件不支持雅加达。jmx连接工厂。

apache-amqp组件是否支持来自Apache Qpid客户端的JMS 2.0？

package org.example;


import jakarta.jms.Connection;
import jakarta.jms.ConnectionFactory;
import jakarta.jms.DeliveryMode;
import jakarta.jms.Destination;
import jakarta.jms.ExceptionListener;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageProducer;
import jakarta.jms.Session;
import jakarta.jms.TextMessage;
import org.apache.camel.component.jms.JmsConfiguration;
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;

import javax.naming.Context;
        import javax.naming.InitialContext;

public class AMQPArtemisClient {
    public static void main(String[] args) throws Exception {
 
        try {
            // The configuration for the Qpid InitialContextFactory has been supplied in
            // a jndi.properties file in the classpath, which results in it being picked
            // up automatically by the InitialContext constructor.
            Context context = new InitialContext();

            ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
            Destination queue = (Destination) context.lookup("myQueueLookup");

            System.setProperty("USER","admin");
            System.setProperty("PASSWORD","admin");

            //added for poolable connection
            JmsPoolConnectionFactory poolConnectionFactory = new JmsPoolConnectionFactory();
            poolConnectionFactory.setMaxConnections(5);
            poolConnectionFactory.setConnectionFactory(factory);

            //  Connection connection = factory.createConnection(System.getProperty("USER"), System.getProperty("PASSWORD"));
            Connection connection = poolConnectionFactory.createConnection(System.getProperty("USER"), System.getProperty("PASSWORD"));
            connection.setExceptionListener(new MyExceptionListener());
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            MessageProducer messageProducer = session.createProducer(queue);
            MessageConsumer messageConsumer = session.createConsumer(queue);

            TextMessage message = session.createTextMessage("Hello world!");
            messageProducer.send(message, DeliveryMode.NON_PERSISTENT, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
            TextMessage receivedMessage = (TextMessage) messageConsumer.receive(2000L);

            if (receivedMessage != null) {
                System.out.println(receivedMessage.getText());
            } else {
                System.out.println("No message received within the given timeout!");
            }

            connection.close();

        } catch (Exception exp) {
            System.out.println("Caught exception, exiting.");
            exp.printStackTrace(System.out);
            System.exit(1);
        }
    }

    private static class MyExceptionListener implements ExceptionListener {
        @Override
        public void onException(JMSException exception) {
            System.out.println("Connection ExceptionListener fired, exiting.");
            exception.printStackTrace(System.out);
            System.exit(1);
        }
    }
}

• 资源/jndi.properties

java.naming.factory.initial = org.apache.qpid.jms.jndi.JmsInitialContextFactory
connectionfactory.myFactoryLookup = amqp://localhost:5672
queue.myQueueLookup = queue
topic.myTopicLookup = topic

• pom.xml

    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-amqp</artifactId> <!--version 3.17.0 is used-->
    </dependency>

<dependency>
          <groupId>org.messaginghub</groupId>
          <artifactId>pooled-jms</artifactId>
          <version>3.0.0</version>  <!-- supports jakarta.jms connecton factory -->
      </dependency>
      <dependency>
          <groupId>org.apache.qpid</groupId>
          <artifactId>qpid-jms-client</artifactId>
          <version>2.0.0</version>
      </dependency>

• Camel组件我试图配置AMPQ组件

   <bean id="jmsConnectionFactory" class="org.apache.qpid.jms.JmsConnectionFactory" >
        <property name="username" value="admin"/>
        <property name="password" value="secret"/>
       <property name="remoteURI" value="amqp://localhost:5672" />
    </bean>

   <bean id="jmsPooledConnectionFactory" class="org.messaginghub.pooled.jms.JmsPoolConnectionFactory" init-method="start" destroy-method="stop">
        <property name="maxConnections" value="5" />
        <property name="connectionFactory" ref="jmsConnectionFactory" />
    </bean>


    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="jmsPooledConnectionFactory" />
        <property name="concurrentConsumers" value="5" />
    </bean>
    <!-- uses the  AMQP component -->
   <bean id="jms" class="org.apache.camel.component.amqp.AMQPComponent">
        <property name="configuration" ref="jmsConfig" />
       <property name="connectionFactory" ref="jmsPooledConnectionFactory"/>
    </bean>

当我尝试下面这样的配置时，我得到异常雅加达。jms期望javax的异常。jms，现在我不得不将消息传递中心的版本降低到2.0.5，并使用qpid jms客户端jar到1.6.0。

<bean id="amqp" class="org.apache.camel.component.amqp.AmqpComponent">
   <property name="connectionFactory">
     <bean class="org.apache.qpid.jms.JmsConnectionFactory" factory-method="createFromURL">
       <property name="remoteURI" value="amqp://localhost:5672" />
       <property name="topicPrefix" value="topic://" />  <!-- only necessary when connecting to ActiveMQ over AMQP 1.0 -->
     </bean>
   </property>
 </bean>

Answer 1

您正在使用的Camel AMQP组件不支持Jakarta JMS an，因此您必须恢复到Qpid JMS客户端和池JMS包装器的旧版本，才能使其工作。这是您唯一的选择，直到有一个支持雅加达的驼峰比特版本，然后您继续。