原文鏈接：https://blog.csdn.net/My_harbor/article/details/81328727
一、安裝ActiveMq
具體安裝步驟：自己谷歌去
二、新建springboot項目
具體步驟：自己谷歌去
三、項目結構
四、引入activemq
<!-- activeMq消息隊列 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
<version>1.4.0.RELEASE</version>
</dependency>
五、編寫消息數據模型
1、JMS定義了五種不同的消息格式，可以根據實際情況選則使用，本例子使用的是ObjectMessage（序列化的Java對象）
StreamMessage -- Java原始值的數據流
· MapMessage--一套名稱-值對
· TextMessage--一個字符串對象
· ObjectMessage--一個序列化的 Java對象
· BytesMessage--一個未解釋字節的數據流
2、編寫的消息對象需要實現序列化接口，代碼如下：
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") +
'}';
}
}
六、編寫配置類
從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;
}
}
七、編寫生產者
生產者提供兩個發送消息的方法，一個是即時發送消息，一個是延時發送消息。延時發送消息需要手動修改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();
}
}
}
}
八、編寫消費者，監聽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());
}
}
九、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();
}
}
}
及時發送的效果如下：
延遲發送效果如下：
