Spring Boot系列——7步整合RabbitMQ
RabbitMQ是一種我們經常使用的訊息中介軟體,通過RabbitMQ可以幫助我們實現非同步、削峰的目的。
今天這篇,我們來看看Spring Boot是如何整合RabbitMQ,傳送訊息和消費訊息的。同時我們介紹下死信佇列。
整合RabbitMQ
整合RabbitMQ只需要如下幾步即可
1、新增maven依賴
<!--rabbitmq--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
2、新增配置檔案application.yaml
在application.yaml新增配置內容如下
spring:rabbitmq: host: 192.168.1.161 port: 5672 username: guest password: guest cache: channel: size: 10 listener: type: simple simple: acknowledge-mode: auto concurrency: 5 default-requeue-rejected: true max-concurrency: 100 retry: enabled: true #initial-interval: 1000ms max-attempts: 3 #max-interval: 1000ms multiplier: 1 stateless: true #publisher-confirms: true</pre>
注意:
這裡最基本的配置只需要配置 host
, port
, username
和 password
四個屬性即可
其他屬性都有各自的含義,比如 retry
是用於配置重試策略的, acknowledge-mode
是配置訊息接收確認機制的。
3、編寫配置類
編寫RabbitConfig配置類,採用Java Configuration的方式配置RabbitTemplate、Exchange和Queue等資訊,具體如下所示
package com.jackie.springbootdemo.config; import org.springframework.amqp.core.*; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; import java.util.HashMap; import java.util.Map; @Configuration public class RabbitMQConfig implements InitializingBean {@Autowired SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory; @Override public void afterPropertiesSet() throws Exception { simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter()); }@Bean("jackson2JsonMessageConverter") public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ConnectionFactory connectionFactory) { return new Jackson2JsonMessageConverter(); }@Bean("rabbitTemplate") @Scope(ConfigurableBeanFactory.SCOPE_SINGLETON) public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, @Qualifier("jackson2JsonMessageConverter") Jackson2JsonMessageConverter jackson2JsonMessageConverter) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMessageConverter(new Jackson2JsonMessageConverter()); return template; }// --------------------- 宣告佇列 ------------------------ @Bean public Queue demoQueue() { return new Queue("demo_queue"); }// --------------------- 宣告exchange ------------------------@Bean public DirectExchange demoExchange() { return new DirectExchange("demo_exchange"); }// --------------------- 佇列繫結 ------------------------ @Bean public Binding bindingAlbumItemCreatedQueue(DirectExchange demoExchange, Queue demoQueue) { return BindingBuilder.bind(demoQueue).to(demoExchange).with("100"); }}
注意
這裡聲明瞭Direct模式的Exchange,宣告一個Queue,並通過 routing-key
為100將 demo_queue
繫結到 demo_exchange
,這樣 demo_queue
就可以接收到 demo_exchange
傳送的訊息了。
4、編寫訊息傳送類
package com.jackie.springbootdemo.message; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class Sender implements RabbitTemplate.ConfirmCallback {private RabbitTemplate rabbitTemplate; /** * 構造方法注入 */ @Autowired public Sender(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果為單例的話,那回調就是最後設定的內容 }public void sendMsg(String content) { rabbitTemplate.convertAndSend("demo_exchange", "100", content); }/** * 回撥 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(" 回撥id:" + correlationData); if (ack) { System.out.println("訊息成功消費"); } else { System.out.println("訊息消費失敗:" + cause); } }}
注意
傳送內容 content
,路由到 routing-key
為100上,則我們就可以在 demo_queue
佇列中看到傳送的訊息內容了
confirm函式是回撥函式,這裡因為沒有消費者,且 acknoledge-mode
是auto(其他兩種值分別是none和manual),所以ack是false。
5、編寫傳送訊息測試類
package com.jackie.springbootdemo; import com.jackie.springbootdemo.message.Sender; 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 org.springframework.test.context.web.WebAppConfiguration; @RunWith(SpringRunner.class) @SpringBootTest(classes = SpringbootDemoApplication.class) @WebAppConfiguration public class RabbitApplicationTests {@Autowired Sender sender; @Test public void contextLoads() throws Exception { sender.sendMsg("test"); } }
執行該測試類,我們可以看到如下結果
6、編寫訊息消費類
package com.jackie.springbootdemo.message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class Receiver {@RabbitListener(queues = "demo_queue") public void created(String message) { System.out.println("orignal message: " + message); }}
注意
訊息消費類也非常簡單,添加註解@RabbitListener,指定要監聽的佇列名稱即可
除了註解@RabbitListener,我們經常還能看到@RabbitHandler,這兩個註解可以配合起來使用。
@RabbitListener 標註在類上面表示當有收到訊息的時候,就交給 @RabbitHandler 的方法處理,具體使用哪個方法處理,根據 MessageConverter 轉換後的引數型別,形如
@RabbitListener(queues = "demo_queue")public class Receiver {@RabbitHandlerpublic void processMessage1(String message) { System.out.println(message); }@RabbitHandler public void processMessage2(byte[] message) { System.out.println(new String(message)); } }
7、執行訊息傳送測試類
從執行結果可以看到,因為有了消費者,所以這次列印的結果是"訊息消費成功"
而且,我們看到Receiver類將訊息消費並打印出訊息的內容為"test"。
程式碼已經提交至專案 ofollow,noindex" target="_blank">rome :https://github.com/DMinerJackie/rome
本來準備再說說死信佇列的,限於篇幅,後面再寫吧。
如果您覺得閱讀本文對您有幫助,請點一下“ 推薦 ”按鈕,您的“ 推薦 ”將是我最大的寫作動力!如果您想持續關注我的文章,請掃描二維碼,關注JackieZheng的微信公眾號,我會將我的文章推送給您,並和您一起分享我日常閱讀過的優質文章。
