初识activemq（pub/sub模式）

男娘i 2022-03-30 08:56 282阅读 0赞

### 初识activemq（pub/sub模式） ###

*  Subscriber
 *  Listener1
 *  Publisher

> 发布、订阅模式，想象生活中订阅报纸或者关注的公众号发布消息

注意：在pub/sub中，必须先启动sub，不然的话，在启动sub之前发布的消息是不能消费的，就像你今天订开始报纸，那今天之前的报纸你肯定是收不到了，一样的。

> pub/sub和p2p模式基本一样，变化的只有Queue和topic

# Subscriber #

public class Subscriber {
        //用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //url
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        private static ConnectionFactory factory = null;
        private static Connection connection = null;
        private static Session session = null;
    
        public static void main(String[] args) throws JMSException {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,URL);
            connection = factory.createConnection();
            connection.start();
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    
            Destination topic = session.createTopic("topic1");
    
            MessageConsumer subscriber1 = session.createConsumer(topic);
            MessageConsumer subscriber2 = session.createConsumer(topic);
    
            subscriber1.setMessageListener(new Listener1());
            subscriber2.setMessageListener(new Listener1());
    
    
        }
    }

# Listener1 #

public class Listener1 implements MessageListener {
    
        @Override
        public void onMessage(Message message) {
            try {
                System.out.println("订阅者一收到的消息：" + ((TextMessage) message).getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

# Publisher #

public class Publisher {
        //默认连接用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //默认连接密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //默认的连接地址
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
        //发送的消息数量
        private static final int SENNUM = 10;
    
        public static void main(String[] args){
            ConnectionFactory factory ; //连接工厂
            Connection connection = null ; //连接
            Session session ; //会话，接收或者发送消息的线程
            Destination destination; //消息的目的地
            MessageProducer messageProducer; //消息生产者
            //实例化连接工厂
            factory = new ActiveMQConnectionFactory(Publisher.USERNAME, Publisher.PASSWORD, Publisher.BROKEURL);
            //通过连接工厂获取connection
            try {
                connection = factory.createConnection();
                connection.start(); //启动连接
                //创建session
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                //创建消息队列
    
                //创建主题
                destination = session.createTopic("topic1");
                //创建消息发布者
                messageProducer = session.createProducer(destination);
                //发送消息
                sendMessage(session, messageProducer);
                session.commit();
            } catch (JMSException e) {
                e.printStackTrace();
            }finally{
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        /**
         * 发送消息
         * @param session
         * @param mp
         * @throws JMSException
         */
        public static void sendMessage(Session session, MessageProducer mp) throws JMSException{
            for(int i = 0;i<Publisher.SENNUM;i++){
                TextMessage message = session.createTextMessage("ActiveMq 发布的消息" + i);
                System.out.println("发布消息：" + "ActiveMq 发布的消息" + i);
                mp.send(message);
            }
        }
    
    }

需要注意的是，在这个例子中，Publisher发布了10条消息，但是通过activemq的控制台会发现，一共消费了20条消息，这是因为有两个消费者。就像两个人订了同一份报纸，那出版社一定会生产这种报纸两张，而不是一张。

![在这里插入图片描述][20190105153418886.png]

[20190105153418886.png]: /images/20220330/426f8ceee1c243b785a37d05f7379895.png