JMS 消费多个主题

Question

我是 Java 新手，正在从事一个使用多个（不同）主题并将其发送到另一台服务器的项目。我想知道处理多个主题的最佳方式是什么。

据我了解，每个消费者都与一个主题相关联，因此如果我必须使用多个主题，我将需要一个消费者来处理每个不同的主题。由于消费者进行了阻塞调用，我需要为每个消费者调用一个线程来并行使用这些主题。

如果我想进一步提高吞吐量，每个消费者（附加到主题）有一个老板线程并允许每个老板线程设置工作线程以相应地提高性能是一种好习惯吗？

请告知这是否是一种好的做法，如果不是，还有哪些其他替代选择？是否有任何众所周知的设计模式来处理这个问题

为什么我选择了消费者模型而不是侦听器模型？

我还有一个约束，即消费者收到消息后，需要将消息发送到另一台接收服务器。如果接收服务器关闭（在新版本推送期间），那么我必须暂停消费消息，直到接收服务器启动。在这种情况下，拥有消息监听器将无济于事，因为当接收服务器关闭时我将无法暂停监听器。我这样说是对的，还是有办法暂停侦听器并停止消费消息，直到接收服务器启动？

Answer 1

我的方法是使用监听器功能。

您的对象实现了MessageListener 接口，然后您将消息侦听器添加到消费者。在这种情况下，客户端库将为您处理线程，从队列中读取消息并将它们发送给侦听器。

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MyMessageConsumer implements MessageListener {

    public static void main() {
        try {

            MyMessageConsumer myMessageConsumer = new MyMessageConsumer();

            // This example is using the ActiveMQ client library
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("nio://localhost:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();

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

            Destination destination1 = session.createQueue("MyTopic1");
            MessageConsumer consumer1 = session.createConsumer(destination1);
            consumer1.setMessageListener(myMessageConsumer);

            Destination destination2 = session.createQueue("MyTopic2");
            MessageConsumer consumer2 = session.createConsumer(destination2);
            consumer2.setMessageListener(myMessageConsumer);

        } catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        }
    }

    @Override
    public void onMessage(Message message) {
        // Handle my messages here
    }
}

会话交易

在此选项中，我们使用事务处理消息，如果调用 session.rollback()，它将传递消息。当你的操作成功时你确认() 或者当失败时回滚()。

包 io.bessel.test;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MyMessageConsumer implements MessageListener {

    public static void main(String ... arguments) {
        try {
            // This example is using the ActiveMQ client library
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("nio://localhost:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);

            MyMessageConsumer myMessageConsumer = new MyMessageConsumer(session);

            Destination destination1 = session.createQueue("MyTopic1");
            MessageConsumer consumer1 = session.createConsumer(destination1);
            consumer1.setMessageListener(myMessageConsumer);

            Destination destination2 = session.createQueue("MyTopic2");
            MessageConsumer consumer2 = session.createConsumer(destination2);
            consumer2.setMessageListener(myMessageConsumer);

        } catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        }
    }

    private final Session session;

    public MyMessageConsumer(Session session) {
        this.session = session;
    }

    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            try {
                String text = ((TextMessage) message).getText();
                System.out.println(String.format("Received message: %s", text));
                this.session.rollback();

            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

}