Spring Boot 入門之消息中間件的使用
更新時間：2018年02月08日 14:32:08 作者：moonlightL
本篇文章主要介紹了Spring Boot 入門之消息中間件的使用，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
一、前言
在消息中間件中有 2 個重要的概念：消息代理和目的地。當(dāng)消息發(fā)送者發(fā)送消息后，消息就被消息代理接管，消息代理保證消息傳遞到指定目的地。
我們常用的消息代理有 JMS 和 AMQP 規(guī)范。對應(yīng)地，它們常見的實現(xiàn)分別是 ActiveMQ 和 RabbitMQ。
二、整合 ActiveMQ
2.1 添加依賴
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-activemq</artifactId> </dependency> <!-- 如果需要配置連接池，添加如下依賴 --> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> </dependency>
2.2 添加配置
# activemq 配置 spring.activemq.broker-url=tcp://192.168.2.12:61616 spring.activemq.user=admin spring.activemq.password=admin spring.activemq.pool.enabled=false spring.activemq.pool.max-connections=50 # 使用發(fā)布/訂閱模式時，下邊配置需要設(shè)置成 true spring.jms.pub-sub-domain=false
此處 spring.activemq.pool.enabled=false，表示關(guān)閉連接池。
2.3 編碼
配置類：
@Configuration
public class JmsConfirguration {
public static final String QUEUE_NAME = "activemq_queue";
public static final String TOPIC_NAME = "activemq_topic";
@Bean
public Queue queue() {
return new ActiveMQQueue(QUEUE_NAME);
}
@Bean
public Topic topic() {
return new ActiveMQTopic(TOPIC_NAME);
}
}
負(fù)責(zé)創(chuàng)建隊列和主題。
消息生產(chǎn)者：
@Component
public class JmsSender {
@Autowired
private Queue queue;
@Autowired
private Topic topic;
@Autowired
private JmsMessagingTemplate jmsTemplate;
public void sendByQueue(String message) {
this.jmsTemplate.convertAndSend(queue, message);
}
public void sendByTopic(String message) {
this.jmsTemplate.convertAndSend(topic, message);
}
}
消息消費者：
@Component
public class JmsReceiver {
@JmsListener(destination = JmsConfirguration.QUEUE_NAME)
public void receiveByQueue(String message) {
System.out.println("接收隊列消息:" + message);
}
@JmsListener(destination = JmsConfirguration.TOPIC_NAME)
public void receiveByTopic(String message) {
System.out.println("接收主題消息:" + message);
}
}
消息消費者使用 @JmsListener 注解監(jiān)聽消息。
2.4 測試
@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsTest {
@Autowired
private JmsSender sender;
@Test
public void testSendByQueue() {
for (int i = 1; i < 6; i++) {
this.sender.sendByQueue("hello activemq queue " + i);
}
}
@Test
public void testSendByTopic() {
for (int i = 1; i < 6; i++) {
this.sender.sendByTopic("hello activemq topic " + i);
}
}
}
打印結(jié)果：
接收隊列消息:hello activemq queue 1
接收隊列消息:hello activemq queue 2
接收隊列消息:hello activemq queue 3
接收隊列消息:hello activemq queue 4
接收隊列消息:hello activemq queue 5
測試發(fā)布/訂閱模式時，設(shè)置 spring.jms.pub-sub-domain=true
接收主題消息:hello activemq topic 1
接收主題消息:hello activemq topic 2
接收主題消息:hello activemq topic 3
接收主題消息:hello activemq topic 4
接收主題消息:hello activemq topic 5
三、整合 RabbitMQ
3.1 添加依賴
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
3.2 添加配置
spring.rabbitmq.host=192.168.2.30 spring.rabbitmq.port=5672 spring.rabbitmq.username=light spring.rabbitmq.password=light spring.rabbitmq.virtual-host=/test
3.3 編碼
配置類：
@Configuration
public class AmqpConfirguration {
//=============簡單、工作隊列模式===============
public static final String SIMPLE_QUEUE = "simple_queue";
@Bean
public Queue queue() {
return new Queue(SIMPLE_QUEUE, true);
}
//===============發(fā)布/訂閱模式============
public static final String PS_QUEUE_1 = "ps_queue_1";
public static final String PS_QUEUE_2 = "ps_queue_2";
public static final String FANOUT_EXCHANGE = "fanout_exchange";
@Bean
public Queue psQueue1() {
return new Queue(PS_QUEUE_1, true);
}
@Bean
public Queue psQueue2() {
return new Queue(PS_QUEUE_2, true);
}
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(FANOUT_EXCHANGE);
}
@Bean
public Binding fanoutBinding1() {
return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding2() {
return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
}
//===============路由模式============
public static final String ROUTING_QUEUE_1 = "routing_queue_1";
public static final String ROUTING_QUEUE_2 = "routing_queue_2";
public static final String DIRECT_EXCHANGE = "direct_exchange";
@Bean
public Queue routingQueue1() {
return new Queue(ROUTING_QUEUE_1, true);
}
@Bean
public Queue routingQueue2() {
return new Queue(ROUTING_QUEUE_2, true);
}
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE);
}
@Bean
public Binding directBinding1() {
return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
}
@Bean
public Binding directBinding2() {
return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
}
//===============主題模式============
public static final String TOPIC_QUEUE_1 = "topic_queue_1";
public static final String TOPIC_QUEUE_2 = "topic_queue_2";
public static final String TOPIC_EXCHANGE = "topic_exchange";
@Bean
public Queue topicQueue1() {
return new Queue(TOPIC_QUEUE_1, true);
}
@Bean
public Queue topicQueue2() {
return new Queue(TOPIC_QUEUE_2, true);
}
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(TOPIC_EXCHANGE);
}
@Bean
public Binding topicBinding1() {
return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
}
@Bean
public Binding topicBinding2() {
return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
}
}
RabbitMQ 有多種工作模式，因此配置比較多。想了解相關(guān)內(nèi)容的讀者可以查看《RabbitMQ 工作模式介紹》或者自行百度相關(guān)資料。
消息生產(chǎn)者：
@Component
public class AmqpSender {
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 簡單模式發(fā)送
*
* @param message
*/
public void simpleSend(String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
}
/**
* 發(fā)布/訂閱模式發(fā)送
*
* @param message
*/
public void psSend(String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
}
/**
* 路由模式發(fā)送
*
* @param message
*/
public void routingSend(String routingKey, String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
}
/**
* 主題模式發(fā)送
*
* @param routingKey
* @param message
*/
public void topicSend(String routingKey, String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
}
}
消息消費者：
@Component
public class AmqpReceiver {
/**
* 簡單模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
public void simpleReceive(String message) {
System.out.println("接收消息:" + message);
}
/**
* 發(fā)布/訂閱模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
public void psReceive1(String message) {
System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
}
@RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
public void psReceive2(String message) {
System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
}
/**
* 路由模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
public void routingReceive1(String message) {
System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
}
@RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
public void routingReceive2(String message) {
System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
}
/**
* 主題模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
public void topicReceive1(String message) {
System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
}
@RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
public void topicReceive2(String message) {
System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
}
}
消息消費者使用 @RabbitListener 注解監(jiān)聽消息。
3.4 測試
@RunWith(SpringRunner.class)
@SpringBootTest
public class AmqpTest {
@Autowired
private AmqpSender sender;
@Test
public void testSimpleSend() {
for (int i = 1; i < 6; i++) {
this.sender.simpleSend("test simpleSend " + i);
}
}
@Test
public void testPsSend() {
for (int i = 1; i < 6; i++) {
this.sender.psSend("test psSend " + i);
}
}
@Test
public void testRoutingSend() {
for (int i = 1; i < 6; i++) {
this.sender.routingSend("order", "test routingSend " + i);
}
}
@Test
public void testTopicSend() {
for (int i = 1; i < 6; i++) {
this.sender.topicSend("user.add", "test topicSend " + i);
}
}
}
測試結(jié)果略過。。。
踩坑提醒1：ACCESS_REFUSED – Login was refused using authentication mechanism PLAIN
解決方案：
1) 請確保用戶名和密碼是否正確，需要注意的是用戶名和密碼的值是否包含空格或制表符（筆者測試時就是因為密碼多了一個制表符導(dǎo)致認(rèn)證失?。?/p>
2) 如果測試賬戶使用的是 guest，需要修改 rabbitmq.conf 文件。在該文件中添加 “l(fā)oopback_users = none” 配置。
踩坑提醒2：Cannot prepare queue for listener. Either the queue doesn't exist or the broker will not allow us to use it
解決方案：
我們可以登陸 RabbitMQ 的管理界面，在 Queue 選項中手動添加對應(yīng)的隊列。
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了IDEA社區(qū)版下載安裝流程詳解(小白篇)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-08-08
這篇文章主要介紹了使用@CacheEvict清除指定下所有緩存，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
這篇文章主要介紹了Java中的Future獲取任務(wù)返回值詳解,在向線程池ThreadPoolExecutor提交任務(wù)時,一般為了方便操作采用execute提交任務(wù),這時線程其實是無返回值的,需要的朋友可以參考下2023-12-12
這篇文章主要給大家介紹了如何通過三分鐘學(xué)會Java枚舉的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
提交gRPC-spring-boot-starter項目bug修復(fù)的pr說明
這篇文章主要介紹了這篇文章主要為大家介紹了gRPC-spring-boot-starter項目提交bug修復(fù)的pr的原因說明，有需要的朋友可以借鑒參考下，希望能夠有所幫助2022-02-02
在我們平時的代碼過程中，鏈表是我們經(jīng)常遇到的一個數(shù)據(jù)結(jié)構(gòu)，它非常的簡單，但Java并不能直接將一個鏈表打印出來，通過這篇文章我們來講解一下這個問題2022-01-01
Spring?Boot?3中一套可以直接用于生產(chǎn)環(huán)境的Log4J2日志配置詳解
Log4J2是Apache Log4j的升級版，參考了logback的一些優(yōu)秀的設(shè)計，并且修復(fù)了一些問題，因此帶來了一些重大的提升，這篇文章主要介紹了Spring?Boot?3中一套可以直接用于生產(chǎn)環(huán)境的Log4J2日志配置,需要的朋友可以參考下2023-12-12
MyBatis是一個持久層框架ORM框架,實現(xiàn)數(shù)據(jù)庫中數(shù)據(jù)和Java對象中的屬性的雙向映射,那么不可避免的就會碰到類型轉(zhuǎn)換的問題,本文主要介紹了MyBatis類型轉(zhuǎn)換模塊的實現(xiàn),感興趣的可以了解一下2023-09-09
這篇文章主要介紹了Java程序命令行參數(shù)用法總結(jié)，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
SpringBoot+EasyPoi實現(xiàn)excel導(dǎo)出功能
最新小編遇到這樣一個需求，根據(jù)檢索條件查詢列表并將結(jié)果導(dǎo)出到excel，實現(xiàn)過程也非常簡單，感興趣的朋友跟隨小編一起看看吧2021-09-09