SpringBoot集成ActiveMq消息队列实现消息即时处理、延迟处理

 

1、安装ActiveMq

具体安装步骤：本身谷歌去

2、新建springboot项目

具体步骤：本身谷歌去

3、项目结构

4、引入activemq

<!-- activeMq消息队列 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
            <version>1.4.0.RELEASE</version>
        </dependency>

5、编写消息数据模型

一、JMS定义了五种不一样的消息格式，能够根据实际状况选则使用，本例子使用的是ObjectMessage（序列化的Java对象）

 StreamMessage -- Java原始值的数据流

· MapMessage--一套名称-值对

· TextMessage--一个字符串对象

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

· BytesMessage--一个未解释字节的数据流

二、编写的消息对象须要实现序列化接口，代码以下：

package com.sky.frame.activemq;

import cn.hutool.core.date.DateUtil;
import lombok.*;

import java.io.Serializable;
import java.util.Date;

/**
 * 消息模型
 * 消息能够是任意数据类型
 */
@Getter
@Setter
@Builder
public class MessageModel implements Serializable {

    private String titile;
    private String message;

    @Override
    public String toString() {
        return "MessageModel{" +
                "titile='" + titile + '\'' +
                ", message='" + message + '\'' +
                ", date=" + DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss") +
                '}';
    }
}

6、编写配置类

从ActiveMQ5.12.2开始，为了加强安全行，ActiveMQ强制用户配置可序列化的包名，不然报错

package com.sky.frame.activemq.config;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @modified By
 * @description 从ActiveMQ5.12.2 开始，为了加强这个框架的安全性，ActiveMQ将强制用户配置可序列化的包名
 * @date Created in 2018/7/31 19:09
 */
@Component
public class ActiveMqConfig {

    @Bean
    public ActiveMQConnectionFactory factory(@Value("${spring.activemq.broker-url}") String url){
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
        // 设置信任序列化包集合
        List<String> models = new ArrayList<>();
        models.add("java.lang");
        models.add("java.util");
        models.add("com.sky.frame.activemq");
        factory.setTrustedPackages(models);
        // 设置处理机制
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(0); // 消息处理失败从新处理次数
        factory.setRedeliveryPolicy(redeliveryPolicy);
        return factory;
    }
}

7、编写生产者

生产者提供两个发送消息的方法，一个是即时发送消息，一个是延时发送消息。延时发送消息须要手动修改activemq目录conf下的activemq.xml配置文件，开启延时，设置schedulerSupport="true"，而后重启activemq便可

能够在生产者里事先定义好消息队列，我在这里定义好了一个default.queue队列，用于一会测试

package com.sky.frame.activemq;

import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.jms.JmsProperties;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.*;
import java.io.Serializable;


/**
 * 消息生产者
 */
@Service("acProducer")
public class Producer {

    public static final Destination DEFAULT_QUEUE = new ActiveMQQueue("default.queue");

    @Autowired
    private JmsMessagingTemplate template;

    /**
     * 发送消息
     * @param destination destination是发送到的队列
     * @param message message是待发送的消息
     */
    public <T extends Serializable> void send(Destination destination, T message){
        template.convertAndSend(destination, message);
    }

    /**
     * 延时发送
     * @param destination 发送的队列
     * @param data 发送的消息
     * @param time 延迟时间
     */
    public <T extends Serializable> void delaySend(Destination destination, T data, Long time){
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        // 获取链接工厂
        ConnectionFactory connectionFactory = template.getConnectionFactory();
        try {
            // 获取链接
            connection = connectionFactory.createConnection();
            connection.start();
            // 获取session，true开启事务，false关闭事务
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 建立一个消息队列
            producer = session.createProducer(destination);
            producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
            ObjectMessage message = session.createObjectMessage(data);
            //设置延迟时间
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
            // 发送消息
            producer.send(message);
            session.commit();
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            try {
                if (producer != null){
                    producer.close();
                }
                if (session != null){
                    session.close();
                }
                if (connection != null){
                    connection.close();
                }
            } catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

8、编写消费者，监听default.queue这个消息队列

package com.sky.frame.activemq;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 */
@Component
public class Consumer {

    @JmsListener(destination = "default.queue")
    public void receiveQueue(MessageModel message){
        System.out.println("收到消息"+message.toString());
    }
}

9、yml配置文件

好了，到这里springboot整合activeMQ完成了，接下来咱们来测试下，测试的消息往以前定义好的default.queue队列发送，

代码入下：

import com.sky.SkyApplication;
import com.sky.frame.activemq.MessageModel;
import com.sky.frame.activemq.Producer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Date;


/**
 * @modified By
 * @description
 * @date Created in 2018/7/31 17:19
 */
@SpringBootTest(classes = SkyApplication.class)
@RunWith(SpringRunner.class)
public class ActiveMqTest {

    /**
     * 消息生产者
     */
    @Autowired
    private Producer producer;

    /**
     * 及时消息队列测试
     */
    @Test
    public void test(){
        MessageModel messageModel = MessageModel.builder()
                .message("测试消息")
                .titile("消息000")
                .build();
        // 发送消息
        producer.send(Producer.DEFAULT_QUEUE, messageModel);
    }

    /**
     * 延时消息队列测试
     */
    @Test
    public void test2(){
        for (int i=0;i< 20;i++){
            MessageModel messageModel = MessageModel.builder()
                    .titile("延迟10秒执行")
                    .message("测试消息" + i)
                    .build();
            // 发送延迟消息
            producer.delaySend(Producer.DEFAULT_QUEUE, messageModel, 10000L);
        }
        try {
            // 休眠100秒，等等消息执行
            Thread.currentThread().sleep(100000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

及时发送的效果以下：

延迟发送效果以下：

 

好了，到这里结束了，有什么不对的地方，欢迎你们指点，谢谢。