Spring Boot系列——7步整合RabbitMQ
阿新 • • 發佈：2018-12-16
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 { @RabbitHandler public void processMessage1(String message) {
System.out.println(message);
} @RabbitHandler
public void processMessage2(byte[] message) {
System.out.println(new String(message));
} }
7、執行訊息傳送測試類
從執行結果可以看到，因為有了消費者，所以這次列印的結果是"訊息消費成功"
而且，我們看到Receiver類將訊息消費並打印出訊息的內容為"test"。
程式碼已經提交至專案rome：https://github.com/DMinerJackie/rome
本來準備再說說死信佇列的，限於篇幅，後面再寫吧。
如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕，您的“推薦”將是我最大的寫作動力！如果您想持續關注我的文章，請掃描二維碼，關注JackieZheng的微信公眾號，我會將我的文章推送給您，並和您一起分享我日常閱讀過的優質文章。
