ActiveMQ--topic

以你之姓@ 2021-08-14 01:41 331阅读 0赞

### 发布/订阅模式（topic） ###

发布/订阅模式传递域的特点如下：

*  生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系。
 *  生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费**自它订阅之后发布的消息**。
 *  生产者生产时，topic不保存消息，它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者。

JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。类似微信公众号，关注之后可以查看发布过的所有消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

发布主题生产者：

package pres.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class JmsProduce_topic { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String TOPIC_NAME = "topic_acton";
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //创建消息的生产者
            MessageProducer messageProducer = session.createProducer(topic);
            //通过使用messageProducer生产3条消息发送到MQ的队列里面
            for (int i = 0; i < 3; i++) { 
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);//可以理解为字符串
                //8.通过messageProducer发送给MQ
                messageProducer.send(textMessage);
            }
            //9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();;
            System.out.println("消息发布到TOPIC完成！");
        }
    }

主题消费者

package pres.zhang.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    public class JmsConsumer_topic { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String TOPIC_NAME = "topic_acton";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(topic);
            //6.监听消息
            messageConsumer.setMessageListener(message -> { 
                if(message != null && message instanceof TextMessage){ 
                    TextMessage textMessage = (TextMessage) message;
                    try { 
                        System.out.println("监听器接收消息：" + textMessage.getText());
                    } catch (JMSException e) { 
                        e.printStackTrace();
                    }
                }
            });
            //保证控制台运行
            System.in.read();
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

注意：先启动订阅，再启动消费者：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

监听器接收消息：TOPIC_NAME---0
    监听器接收消息：TOPIC_NAME---1
    监听器接收消息：TOPIC_NAME---2

### queue与topic对比 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/aa16047da9da44afa919da55e33e4e86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/7976b70701c2414c92055b89fa453b8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/d7712e684fac43f39e343fd946cabed1.png

ActiveMQ--topic

发表评论取消回复

还没有评论，来说两句吧...

相关阅读