ActiveMQ的使用

叁歲伎倆 2022-02-05 01:23 261阅读 0赞

### 什么情况下使用ActiveMQ? ###

1.  多个项目之间集成   
    (1) 跨平台   
    (2) 多语言   
    (3) 多项目
2.  降低系统间模块的耦合度，解耦   
    (1) 软件扩展性
3.  系统前后端隔离   
    (1) 前后端隔离，屏蔽高安全区

### ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70][] ###

### 生产者代码：  ###

package com.ljxwtl.cn;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class Producer {
    
        //连接账号
        private String userName = ActiveMQConnectionFactory.DEFAULT_USER;
        //连接密码
        private String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
        //连接地址
        private String brokerURL = "tcp://localhost:61616";
        //connection的工厂
        private ConnectionFactory factory;
        //连接对象
        private Connection connection;
        //一个操作会话
        private Session session;
        //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
        private Destination destination;
        //生产者，就是产生数据的对象
        private MessageProducer producer;
    
        public void start() throws Exception{
            //根据用户名，密码，url创建一个连接工厂
            factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
            //从工厂中获取一个连接
            connection = factory.createConnection();
            //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
            connection.start();
            //创建一个session
            //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
            session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
            destination = session.createQueue("text-msg");
            //从session中，获取一个消息生产者
            producer = session.createProducer(null);
            //设置生产者的模式，有两种可选
            //DeliveryMode.PERSISTENT 当activemq关闭的时候，队列数据将会被保存
            //DeliveryMode.NON_PERSISTENT 当activemq关闭的时候，队列里面的数据将会被清空
            //producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            //创建一条消息，当然，消息的类型有很多，如文字，字节，对象等,可以通过session.create..方法来创建出来
            for(int i = 0 ; i < 100 ; i ++){
                TextMessage textMsg = session.createTextMessage("我是一条消息"+i);
                //发送一条消息
                /**
                 * 第一个参数：目的地
                 * 第二个参数：消息
                 * 第三个参数：持久化模式
                 * 第四个参数：优先级（0-9   0-4为普通消息，5-9为加急消息，默认为4）
                 * 第五个参数：消息在MQ上存放的有效期 ms
                 */
                producer.send(destination,textMsg,DeliveryMode.NON_PERSISTENT,5,1000*60*2);
            }
            session.commit();
            System.out.println("发送消息成功");
            //即便生产者的对象关闭了，程序还在运行哦
            connection.close();
        }
    
        public static void main(String[] args) throws Exception {
            new Producer().start();
        }
    }

### 消费者代码： ###

package com.ljxwtl.cn;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class Consumer {
    
        //连接账号
        private String userName = ActiveMQConnectionFactory.DEFAULT_USER;
        //连接密码
        private String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
        //连接地址
        private String brokerURL = "tcp://localhost:61616";
        //connection的工厂
        private ConnectionFactory factory;
        //连接对象
        private Connection connection;
        //一个操作会话
        private Session session;
        //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
        private Destination destination;
        //消费者，就是消费数据的对象
        private MessageConsumer messageConsumer;
    
        public void start() throws Exception{
            //根据用户名，密码，url创建一个连接工厂
            factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
            //从工厂中获取一个连接
            connection = factory.createConnection();
            //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
            connection.start();
            //创建一个session
            //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
            session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
            destination = session.createQueue("text-msg");
            //从session中，获取一个消息生产者
            messageConsumer = session.createConsumer(destination);
    
            //创建一条消息，当然，消息的类型有很多，如文字，字节，对象等,可以通过session.create..方法来创建出来
    
            while(true){
                Message message = messageConsumer.receive();
                //确认签收，对应于Session.CLIENT_ACKNOWLEDGE
                message.acknowledge();
                if(message == null){
                    break;
                }
                System.out.println("接收到的消息："+message);
            }
    
            connection.close();
        }
    
        public static void main(String[] args) throws Exception {
            new Consumer().start();
        }
    }

查看ActiveMQ消息队列转发的网址：[http://localhost:8161/admin][http_localhost_8161_admin] 用户名:admin 密码：admin

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70]: /images/20220205/31dcde4f16394c06942b30428b59b796.png
[http_localhost_8161_admin]: http://localhost:8161/admin