如何在活动MQ5.x中启动线程会话

Question

我想这是一个标准的知识问题，但是current_session是否会以线程形式运行？

ActiveMQSession current_session = (ActiveMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Thread sessionThreads = new Thread(current_session);
sessionThreads.start();

我想要的是与生产者/消费者的并发会话，这些会话编写/监听他们的特定队列。我已经尝试通过将连接传递给线程来编写一个自定义线程，但是当我创建生产者时，我遇到了一个错误，“我不能在未注册的会话上启动生产者”。

Answer 1

java.lang.Thread没有接受org.apache.activeMQ.activeMQsession对象的构造函数。您的代码甚至无法编译，更不用说运行了。

下面是一个简单的客户端,它将创建用于生成和使用消息的线程：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;

public class MyMultiThreadedApp {

   class MyConsumer extends Thread {
      private final Connection connection;
      private final Destination destination;

      MyConsumer(Connection connection, Destination destination) {
         this.connection = connection;
         this.destination = destination;
      }

      @Override
      public void run() {
         try (Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE)) {
            MessageConsumer messageConsumer = session.createConsumer(destination);
            connection.start();
            Message message = messageConsumer.receive(5000);
            if (message == null) {
               System.out.println("Did not receive message within the allotted time.");
               return;
            }
            System.out.println("Received message: " + message);
         } catch (Throwable e) {
            e.printStackTrace();
            return;
         }
      }
   }

   class MyProducer extends Thread {
      private final Connection connection;
      private final Destination destination;

      MyProducer(Connection connection, Destination destination) {
         this.connection = connection;
         this.destination = destination;
      }

      @Override
      public void run() {
         try (Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE)) {
            MessageProducer messageProducer = session.createProducer(destination);
            messageProducer.send(session.createTextMessage("My message"));
            System.out.println("Sent message");
         } catch (Throwable e) {
            e.printStackTrace();
            return;
         }
      }
   }

   public static void main(String... args) throws Exception {
      MyMultiThreadedApp myMultiThreadedApp = new MyMultiThreadedApp();
      InitialContext initialContext = null;   
      initialContext = new InitialContext();
      Queue queue = (Queue) initialContext.lookup("queue/exampleQueue");
      ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("ConnectionFactory");
      Connection connection = cf.createConnection();
      Thread myConsumer = myMultiThreadedApp.runConsumer(connection, queue);
      Thread myProducer = myMultiThreadedApp.runProducer(connection, queue);
      myConsumer.join();
      myProducer.join();
   }

   private Thread runConsumer(Connection connection, Destination destination) {
      MyConsumer myConsumer = new MyConsumer(connection, destination);
      myConsumer.start();
      return myConsumer;
   }

   private Thread runProducer(Connection connection, Destination destination) {
      MyProducer myProducer = new MyProducer(connection, destination);
      myProducer.start();
      return myProducer;
   }
}