activemq的入门及使用方法

雨点打透心脏的1/2处 2022-04-11 10:11 553阅读 0赞

# 1、什么是activemq #

ActiveMQ 是Apache出品，最流行的、能力强劲的***开源消息总线***。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的JMS Provider实现。

主要特点：

1. 多种语言和协议编写客户端。

2. 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去。

3. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

4. 支持Ajax

5. 支持与Axis的整合

6. 可以很容易得调用内嵌JMS provider,进行测试

# 2、ActiveMQ的消息形式 #

对于消息的传递有两种类型：

**一种是点对点的**，即一个生产者和一个消费者一一对应，这种方式基于**queue队列**

**一种是发布/****订阅模式**，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收（一对多，这种方式**基于topic**）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70][]

对于**点对点的消息**传输模型来说，可以在同一个queue上面注册多个生产者和多个消费者，当消息的生产者发送一条消息时，只有其中的一个消息消费者接收到消息生产者发送的消息，而不是所有消息消费者都会接收到该消息。

对于**发布/订阅者**模式来说，消息的发布者需将消息投递给topic,而消息的订阅者则需要在相应的topic进行注册，以便接受相应topic的消息。与点对点传输模型不同的是，消息发布者的消息将被自动发送给所有订阅了该topic的消息订阅者。当消息订阅者某段时间由于某种原因断开了与消息发布者的连接时，这个时间段的消息将会丢失，除非将消息的订阅模式设置为**持久订阅**，这时消息的发布者将会为消息的订阅者保留这段时间所产生的消息。当消息的订阅者重新连接消息发布者时，消息订阅者仍然可以获得这部分消息，而不至于丢失这部分消息。

JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

*　· StreamMessage -- Java原始值的数据流*

*　　· MapMessage--一套名称-值对*

*　　· TextMessage--一个字符串对象*

*　　· ObjectMessage--一个序列化的 Java对象*

*　　· BytesMessage--一个字节的数据流*

# 2、如何安装activemq #

第一步：进入[http://activemq.apache.org/][http_activemq.apache.org]下载ActiveMQ

第二步：将下载好的安装包上传到linux系统中，并解压（tar -zxvf命令）

第三步：启动。 ./activema start

使用bin目录下的activemq命令启动：
    [root@localhost bin]# ./activemq start
    
    关闭：
    [root@localhost bin]# ./activemq stop
    
    查看状态：
    [root@localhost bin]# ./activemq status

第四步：在浏览器中输入[http://你的ip:8161/admin][http_ip_8161_admin]， 用户名：admin， 密码：admin，可以看到activemq的后台管理页面

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 1][]

# 3、activemq的使用 #

在maven项目中导入activemq的依赖包：

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

##    3.1 点对点（queue） ##

###        3.1.1 生产者代码 ###

public void testQueueProducer() throws Exception {
            //第一步：创建connectionFactory, 指定ip和端口
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
            //第二步：通过connectionFactory创建connection
            Connection connection = connectionFactory.createConnection();
            // 第三步：开启连接，调用Connection对象的start方法。
            connection.start();
            // 第四步：使用Connection对象创建一个Session对象。
            //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
            //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
            //参数：队列的名称。
            Destination destination = session.createQueue("test-queue");
            // 第六步：使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(destination);
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            /*TextMessage message = new ActiveMQTextMessage();
            message.setText("hello activemq");*/
            TextMessage message = session.createTextMessage("hello activemq queue");
            // 第八步：使用Producer对象发送消息。
            producer.send(destination, message);
            // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

###        3.1.2 消费者代码 ###

public void testQueueConsumer() throws Exception {
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("test-queue");
            MessageConsumer consumer = session.createConsumer(destination);
    
            //为consumer设置一个监听器
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        TextMessage msg = (TextMessage)message;
                        System.out.println(msg.getText());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
    
    
            //等待键盘输入
            System.in.read();
    
            //关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

首先运行生产者的代码，在后台管理界面看到已经创建了一个叫test-queue的队列，并且有一条的消息入队。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 2][]

再运行消费者代码：可以看到刚才未发送的消息已经出队了，并且在控制台接收到了发送的消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 3][]

![2018120516502684.png][]

##    3.2 发布/订阅模式（topic） ##

###        3.2.1 生产者 ###

public void testTopicProducer() throws Exception {
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
            //第二步：通过connectionFactory创建connection
            Connection connection = connectionFactory.createConnection();
            // 第三步：开启连接，调用Connection对象的start方法。
            connection.start();
            // 第四步：使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步：使用Session对象创建一个Destination对象
            Destination destination = session.createTopic("test-topic");
            // 第六步：使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(destination);
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage message = session.createTextMessage("hello activemq topic");
            // 第八步：使用Producer对象发送消息。
            producer.send(destination, message);
            // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

###        3.2.2 消费者(为了便于测试，设置了如下的3个消费者) ###

public void testTopicConsumer01() throws Exception {
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://你的ip:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("test-topic");
            MessageConsumer consumer = session.createConsumer(destination);
    
            //为consumer设置一个监听器
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        TextMessage msg = (TextMessage)message;
                        System.out.println(msg.getText());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
    
            System.out.println("topic的消费端01。。。。。");
    
            //等待键盘输入
            System.in.read();
    
            //关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

由于点对点这种模式，消息默认是持久化到activemq中的，即使消费者暂时不在线，但是当消费者上线时，还是会收到消息。而发布者/订阅者模式默认则不会持久化，当消费者暂时断线时，则这段时间的消息就会丢失。因此我们先将3个消费者运行起来，再运行生产者。

# 4、activemq与spring整合 #

生产者配置文件：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 4][]

业务逻辑：在添加新的商品时，同时生成该商品的索引。

@Override
        public TaotaoResult addItem(TbItem item, String desc, String itemParams) {
            //生成商品id
            long itemId = IDUtils.genItemId();
            item.setId(itemId);
            //设置商品状态 '商品状态，1-正常，2-下架，3-删除',
            item.setStatus((byte) 1);
            //设置商品创建时间
            Date date = new Date();
            item.setCreated(date);
            item.setUpdated(date);
    
            itemMapper.insert(item);
    
            //商品描述
            TbItemDesc itemDesc = new TbItemDesc();
            itemDesc.setItemId(itemId);
            itemDesc.setItemDesc(desc);
            itemDesc.setCreated(date);
            itemDesc.setUpdated(date);
            itemDescMapper.insert(itemDesc);
    
            //添加一个商品信息后，向activemq的topic中发送商品id
            jmsTemplate.send(topicDestination, new MessageCreator() {
                @Override
                public Message createMessage(Session session) throws JMSException {
                    TextMessage message = session.createTextMessage(itemId + "");
                    return message;
                }
            });
    
            return TaotaoResult.ok();
        }

消费者配置文件：

监听器代码：

public class ItemChangeMessageListener implements MessageListener {
    
        @Autowired
        private SearchItemService searchItemService;
    
        @Override
        public void onMessage(Message message) {
            try {
                TextMessage textMessage = null;
                Long itemId = null;
                if(message instanceof TextMessage) {
                    textMessage = (TextMessage)message;
                    itemId = Long.parseLong(textMessage.getText());
                }
    
                //向索引库添加文档
                searchItemService.addDocument(itemId);
    
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

监听器监听到来自新增商品的id后，在数据库中查询到该商品的详细信息，通过solrJ为该商品新建索引。

@Override
        public TaotaoResult addDocument(long itemId) throws Exception {
            SearchItem searchItem = searchItemMapper.getItemById(itemId);
            SolrInputDocument document = new SolrInputDocument();
            document.addField("id", searchItem.getId());
            document.addField("item_title", searchItem.getTitle());
            document.addField("item_sell_point", searchItem.getSell_point());
            document.addField("item_price", searchItem.getPrice());
            document.addField("item_image", searchItem.getImage());
            document.addField("item_category_name", searchItem.getCategory_name());
            document.addField("item_desc", searchItem.getItem_desc());
            solrServer.add(document);
            solrServer.commit();
            return TaotaoResult.ok();
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70]: /images/20220411/1d912a708243408dafa869046fe147eb.png
[http_activemq.apache.org]: http://activemq.apache.org/
[http_ip_8161_admin]: http://47.100.187.237:8161/admin/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 1]: /images/20220411/1ae0e429d5a7478e9d4aabb831fd7134.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 2]: /images/20220411/c30cfd1d1c3e4fb88268590b945a4b03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 3]: /images/20220411/942a87cd29f042dd8e5c7176f3fc4e3c.png
[2018120516502684.png]: /images/20220411/4f94ce01ef1f43918c907e214353e826.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 4]: /images/20220411/0da6430eef744eadabf0d20da0dd2261.png