AMQ初级使用（队列模式+主题模式）

男娘i 2022-05-22 06:20 214阅读 0赞

# 队列模式    #

1.生产者代码：

package com.zzf.jms.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class AppProducer {
    
        private static final  String url="tcp://192.168.1.6:61616";
        private static final  String queueName="queue-text";
    
        public static void main(String[] args) throws JMSException {
    
            //1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //((ActiveMQConnectionFactory) connectionFactory).setUserName("admin");
            //((ActiveMQConnectionFactory) connectionFactory).setPassword("admin");
    
            //2.创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3.启动连接
            connection.start();
    
            //4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5.创建接受者目标
            Destination destination = session.createQueue(queueName);
    
            //6.创建目标
            MessageProducer producer = session.createProducer(destination);
    
            for (int i = 0; i <10 ; i++) {
                //7.创建消息
                TextMessage textMessage=session.createTextMessage("消息内容"+i);
                //8.发送消息
                producer.send(textMessage);
                //
                System.out.println(textMessage.getText());
            }
    
            //9.关闭连接
            connection.close();
        }
    }

上文的url可以改成“tcp://localhost:61616”,或者通过cmd->ipconfig

2.消费者代码：

package com.zzf.jms.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class AppConsumer {
    
        private static final  String url="tcp://192.168.1.6:61616";
        private static final  String queueName="queue-text";
    
        public static void main(String[] args) throws JMSException {
    
            //1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //((ActiveMQConnectionFactory) connectionFactory).setUserName("admin");
            //((ActiveMQConnectionFactory) connectionFactory).setPassword("admin");
    
            //2.创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3.启动连接
            connection.start();
    
            //4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5.创建目标
            Destination destination = session.createQueue(queueName);
    
            //6.创建一个消费者
            MessageConsumer consumer = session.createConsumer(destination);
    
            //7.创建一个监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("接受消息："+textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
    
                }
            });
    
            //8.关闭连接
           // connection.close();
        }
    
    }

输出（启动消费者模式，再启动生产者）

![70][]

![70 1][]

![70 2][]

（队列模式如上）

# 主题模式    #

1.发布者代码：

package com.zzf.jms.topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class AppProducer {
    
        private static final  String url="tcp://192.168.1.6:61616";
        private static final  String topicName="topic-text";
    
        public static void main(String[] args) throws JMSException {
    
            //1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //((ActiveMQConnectionFactory) connectionFactory).setUserName("admin");
            //((ActiveMQConnectionFactory) connectionFactory).setPassword("admin");
    
            //2.创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3.启动连接
            connection.start();
    
            //4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5.创建接受者目标
            Destination destination = session.createTopic(topicName);
    
            //6.创建目标
            MessageProducer producer = session.createProducer(destination);
    
            for (int i = 0; i <10 ; i++) {
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("消息内容" + i);
                //8.发送消息
                producer.send(textMessage);
                //
                System.out.println(textMessage.getText());
            }
    
            //9.关闭连接
            connection.close();
        }
    }

2.接受者代码：

package com.zzf.jms.topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class AppConsumer {
    
        private static final  String url="tcp://192.168.1.6:61616";
        private static final  String topicName="topic-text";
    
        public static void main(String[] args) throws JMSException {
    
            //1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //((ActiveMQConnectionFactory) connectionFactory).setUserName("admin");
            //((ActiveMQConnectionFactory) connectionFactory).setPassword("admin");
    
            //2.创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3.启动连接
            connection.start();
    
            //4.创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5.创建目标
            Destination destination = session.createTopic(topicName);
    
            //6.创建一个消费者
            MessageConsumer consumer = session.createConsumer(destination);
    
            //7.创建一个监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("接受消息："+textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
    
                }
            });
    
            //8.关闭连接
           // connection.close();
        }
    
    }

唯一需要改的地方是session.createTopic.

![70 3][]

和队列模式一样，先启动接受者（启动一个以上），再启动发布者

![70 4][]

![70 5][]

![70 6][]

每个接受者都接受到了发布者的发布信息（公众号）

注意：需要导入jar包：

<dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.9.0</version>
            </dependency>

[70]: /images/20220522/25f6e42eb2b7486db146e301f86ca7ad.png
[70 1]: /images/20220522/f08279d583624b0ba004e2d0e3503533.png
[70 2]: /images/20220522/53c79b4aff3e4d04b8a7d71017e4e3fc.png
[70 3]: /images/20220522/6171c4b324544eaa932634a664c660a6.png
[70 4]: /images/20220522/7863310c2dd346dba36d1b82d04ed4ba.png
[70 5]: /images/20220522/5b9da7fa43194a6eb5e40c858bfcaa77.png
[70 6]: /images/20220522/6274e1ceab5b4f09ab49ac20a1c2acd1.png