ActiveMQ发布订阅代码实现

ActiveMQ发布/订阅代码实现

1.导入依赖

    org.apache.activemq
    activemq-all
    5.16.2

2.消费者代码

package com.example.queueprj.consumer;

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
import static javax.jms.Session.AUTO_ACKNOWLEDGE;


public class TopicConsumer {
    public static void main(String[] args) throws JMSException, IOException {
        //创建工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建会话
        Session session = connection.createSession(false,AUTO_ACKNOWLEDGE);
        //创建者主题
        Topic topic = session.createTopic("test-topic");
        //创建消费者
        MessageConsumer messageConsumer = session.createConsumer(topic);
        //消费者增加监听器
        messageConsumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.printf("消息输出 = " + textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        //增加键盘输入，防止main方法执行完控制台就结束了进程，正常的web无需此 *** 作
        System.in.read();

        //关闭连接
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

3.消费者执行结果

 

4.生产者代码

package com.example.queueprj.producer;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import static javax.jms.Session.AUTO_ACKNOWLEDGE;


public class TopicProducer {

    public static void main(String[] args) throws JMSException {
        //创建工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建会话
        Session session = connection.createSession(false,AUTO_ACKNOWLEDGE);
        //创建者主题
        Topic topic = session.createTopic("test-topic");
        //创建消息生产者
        MessageProducer messageProducer = session.createProducer(topic);
        //创建消息
        TextMessage textMessage = session.createTextMessage("我是消息体");
        //发送消息
        messageProducer.send(textMessage);

        //关闭连接
        messageProducer.close();
        session.close();
        connection.close();
        System.out.printf("消息发送成功！！！");
    }
}

5.生产者执行结果