activeMq两种实现方式

分手后的思念是犯贱 2021-12-12 09:23 421阅读 0赞

## 第一种：点对点 ##

#发布者
    public class Producer {
        
        private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
        private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
        private static final String brokerURL = "tcp://192.168.178.X:61616";
    
        public static void main(String[] args) throws JMSException {
            //1.创建连接工厂类
            ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
            //2.创建连接
            Connection connection = factory.createConnection();
            //3.启动连接
            connection.start();
            
            //4.创建会话对象session(事务transacted为true,参数2不生效)
            //acknowledgeMode:
            Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
            
            //5.目的地
            Queue queue = session.createQueue("mq-test-01");
            //7.创建发送者
            MessageProducer producer = session.createProducer(queue);
            for(int i=1;i<=10;i++) {
                //6.消息对象
                TextMessage message = session.createTextMessage();
                message.setText("消息"+i);
                //8.发送消息
                producer.send(message);
            }
            //9.会话提交
    //        session.commit();
            
            //10.关闭连接
            connection.close();
        }
    }

#消费者
    public class Consumer1 {
        
        private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
        private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
        private static final String brokerURL = "tcp://192.168.178.X:61616";
    
        public static void main(String[] args) throws JMSException {
            //1.创建连接工厂类
            ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
            //2.创建连接
            Connection connection = factory.createConnection();
            //3.启动连接
            connection.start();
            
            //4.创建会话对象session(事务transacted为true,参数2不生效)
            //acknowledgeMode:
            Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
            
            //5.目的地
            Queue queue = session.createQueue("mq-test-01");
            
            //6.接收消息对象
            MessageConsumer consumer = session.createConsumer(queue);
            
            //7.通过监听器接收消息
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage)message;
                    //获取消息
                    try {
                        String msg = textMessage.getText();
                        System.out.println(msg);
                    } catch (JMSException e) {
                    }
                }
            });
            
        }
    }

## 第二种： 发布者/订阅者 ##

## 启动顺序：先订阅、再发布 ##

#订阅者
    public class Subscriber1 {
        
        private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
        private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
        private static final String brokerURL = "tcp://192.168.129.10:61616";
    
        public static void main(String[] args) throws JMSException {
            //1.创建连接工厂类
            ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
            //2.创建连接
            Connection connection = factory.createConnection();
            //3.启动连接
            connection.start();
            
            //4.创建会话对象session(事务transacted为true,参数2不生效)
            //acknowledgeMode:
            Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
            
            //5.目的地
            Topic topic = session.createTopic("mq-test-02");
            
            //6.接收消息对象
            MessageConsumer consumer = session.createConsumer(topic);
            
            //7.通过监听器接收消息
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage)message;
                    //获取消息
                    try {
                        String msg = textMessage.getText();
                        System.out.println(msg);
                    } catch (JMSException e) {
                    }
                }
            });
            
        }
    }

#发布者
    public class Publisher {
        
        private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
        private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
        private static final String brokerURL = "tcp://192.168.129.10:61616";
    
        public static void main(String[] args) throws JMSException {
            //1.创建连接工厂类
            ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
            //2.创建连接
            Connection connection = factory.createConnection();
            //3.启动连接
            connection.start();
            
            //4.创建会话对象session(事务transacted为true,参数2不生效)
            //acknowledgeMode:
            Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
            
            //5.目的地
            Topic topic = session.createTopic("mq-test-02");
            //7.创建发送者
            MessageProducer producer = session.createProducer(topic);
            for(int i=1;i<=10;i++) {
                //6.消息对象
                TextMessage message = session.createTextMessage();
                message.setText("消息"+i);
                //8.发送消息
                producer.send(message);
                //设置自动持久化
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            }
            //9.会话提交
    //        session.commit();
            
            //10.关闭连接
            connection.close();
        }
    }

转载于:https://www.cnblogs.com/cpp-wth/p/11143876.html