在ActiveMQ中向多个消费者发送消息

Question

我是ActiveMQ的新手.我试图在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

队列语义在所有消费者中一次性传递一次消息.这是根据JMS规范(理解基础知识的一个很好的阅读).

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