在ActiveMQ中向多个使用者发送消息

Question

我是ActiveMQ新手。我曾尝试在activemq中实现生产者-消费者（发送者-接收器）。在我的代码中，我很容易发送

这是我的制片人

MsgProducer。Java语言

package jms_service;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class MsgProducer {

      private static String url = "failover://tcp://localhost:61616";
      public static javax.jms.ConnectionFactory connFactory;
      public static javax.jms.Connection connection;
      public static javax.jms.Session mqSession;
      public static javax.jms.Topic topic;
      public static javax.jms.MessageProducer producer;

      public static void main(String[] args) throws JMSException {

          connFactory = new ActiveMQConnectionFactory(url);  
          connection = connFactory.createConnection("system","manager");
          connection.start(); 
          mqSession = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);  

          topic = mqSession.createTopic("RealTimeData");
          producer = mqSession.createProducer(topic);                  
          producer.setTimeToLive(30000);

          TextMessage message = mqSession.createTextMessage();      

          int seq_id =1;

          while(true)
            {             
                message.setText("Hello world | " +"seq_id #"+seq_id);               
                 producer.send(message);
                 seq_id++;

                 System.out.println("sent_msg =>> "+ message.getText());
               //  if(seq_id>100000) break;

                    try {
                        Thread.sleep(1000);
                        } 
                    catch (InterruptedException e) { e.printStackTrace();}           
              }       

    }

}

MsgConsumer.java

package jms_service;

import java.text.SimpleDateFormat;
import java.util.Calendar;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class MsgConsumer {

          private static String url = "failover://tcp://localhost:61616";     
          public static javax.jms.ConnectionFactory connFactory;
          public static javax.jms.Connection connection;
          public static javax.jms.Session mqSession;
          public static javax.jms.Topic topic;
          public static javax.jms.MessageConsumer consumer;

        public static void main(String[] args) throws JMSException, InterruptedException {

            connFactory = new ActiveMQConnectionFactory(url);
            connection = connFactory.createConnection("system", "manager");
            connection.setClientID("0002");
            //connection.start();               
            mqSession = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
            topic = mqSession.createTopic("RealTimeData");
            consumer = mqSession.createDurableSubscriber(topic, "SUBS01");
            connection.start();

            MessageListener listner = new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage txtmsg = (TextMessage) message;
                            Calendar cal = Calendar.getInstance();
                            //cal.getTime();
                            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
                            String time = sdf.format(cal.getTime());

                            String msg="received_message =>> "+ txtmsg.getText() + " | received_at :: "+time;
                            System.out.println(msg);

                            //consumer.sendData(msg);
                        }

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

            consumer.setMessageListener(listner);  

      }


}

有谁能帮我找出向多个消费者发送信息的方法吗。提前谢谢。

Answer 1

主题是最好的途径。一个生产者对多个消费者或一个出版商对多个订阅者。对于队列，您必须编写一个循环，以获取所有可能的消费者，并使用不同的目的地发送消息。你的动机还将决定是否使用队列或主题。

1. 如果您认为您的消费者可能离线或有网络问题，请选择队列。在这种情况下，当他们重新打开时，他们将收到待处理的消息
2. 对于主题，当断开连接时，它们将无法接收消息，除非您显式保留消息，但新消息会覆盖它们

Answer 2

假设你的问题是

有人能帮助找出向多个消费者发送消息的方法吗

在不阅读完整代码的情况下，一种方法可能是将客户放入一个集合中

static Vector<consumer> vecConsumer;

在这里，您输入每个新客户，并提供所有现有客户的参考。广播就像发送到单个客户端，例如封装在foreach循环中

for(consumer cons : vecConsumer)
{
      //send stuff or put in sending queue
}

Answer 3

队列语义在所有消费者中只传递一次消息。这是根据JMS规范编写的（这是一本了解基础知识的好书）。

主题语义学向每个消费者传递信息。因此，主题可能是您需求的答案。