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

我是ActiveMQ的新手 . 我试图在activemq中实现 生产环境 者 - 消费者（发送者 - 接收者） . 在我的代码中，我可以通过ActiveMQ轻松地从单个 生产环境 者向单个消费者发送和接收消息 . 但是，问题是，我无法将消息发送给同一个制作人的多个消费者 .

这是我的 producer 和 consumer 课程 .

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);  

      }


}

任何人都可以帮助找出向多个消费者发送消息的方式 . 提前致谢 .