三分鐘實現 Spring Boot 集成 RabbitMQ，實現消息隊列服務
消息中間件在互聯網公司使用的越來越多，主要用于在分布式系統中存儲轉發(fā)消息，在易用性、擴展性、高可用性等方面表現不俗。消息隊列實現系統之間的雙向解耦，生產者往消息隊列中發(fā)送消息，消費者從隊列中拿取消息并處理，生產者不用關心是誰來消費，消費者不用關心誰在生產消息，從而達到系統解耦的目的，也大大提高了系統的高可用性和高并發(fā)能力。
接下來介紹Spring Boot對RabbitMQ的支持。如何在SpringBoot項目中使用RabbitMQ？
Spring Boot提供了spring-bootstarter-amqp組件對消息隊列進行支持，使用非常簡單，僅需要非常少的配置即可實現完整的消息隊列服務。
一、Spring Boot集成RabbitMQ
Spring Boot提供了spring-boot-starter-amqp組件，只需要簡單的配置即可與Spring Boot無縫集成。下面通過示例演示集成RabbitMQ實現消息的接收和發(fā)送。
第一步，配置pom包。
創(chuàng)建Spring Boot項目并在pom.xml文件中添加spring-bootstarter-amqp等相關組件依賴：
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-amqp</artifactId>
- </dependency>
在上面的示例中，引入Spring Boot自帶的amqp組件spring-bootstarter-amqp。
第二步，修改配置文件。
修改application.properties配置文件，配置rabbitmq的host地址、端口以及賬戶信息。
- spring.rabbitmq.host=10.2.1.231
- spring.rabbitmq.port=5672
- spring.rabbitmq.username=zhangweizhong
- spring.rabbitmq.password=weizhong1988
- spring.rabbitmq.virtualHost=order
在上面的示例中，主要配置RabbitMQ服務的地址。RabbitMQ配置由spring.rabbitmq.*配置屬性控制。virtual-host配置項指定RabbitMQ服務創(chuàng)建的虛擬主機，不過這個配置項不是必需的。
第三步，創(chuàng)建消費者。
消費者可以消費生產者發(fā)送的消息。接下來創(chuàng)建消費者類Consumer，并使用@RabbitListener注解來指定消息的處理方法。示例代碼如下：
- @Component
- public class Consumer {
- @RabbitHandler
- @RabbitListener(queuesToDeclare = @Queue("rabbitmq_queue"))
- public void process(String message) {
- System.out.println("消費者消費消息111=====" + message);
- }
- }
在上面的示例中，Consumer消費者通過@RabbitListener注解創(chuàng)建偵聽器端點，綁定rabbitmq_queue隊列。
（1）@RabbitListener注解提供了@QueueBinding、@Queue、@Exchange等對象，通過這個組合注解配置交換機、綁定路由并且配置監(jiān)聽功能等。
（2）@RabbitHandler注解為具體接收的方法。
第四步，創(chuàng)建生產者。
生產者用來產生消息并進行發(fā)送，需要用到RabbitTemplate類。與之前的RedisTemplate類似，RabbitTemplate是實現發(fā)送消息的關鍵類。示例代碼如下：
- @Component
- public class Producer {
- @Autowired
- private RabbitTemplate rabbitTemplate;
- public void produce() {
- String message = new Date() + "Beijing";
- System.out.println("生產者產生消息=====" + message);
- rabbitTemplate.convertAndSend("rabbitmq_queue", message);
- }
- }
如上面的示例所示，RabbitTemplate提供了 convertAndSend方法發(fā)送消息。convertAndSend方法有routingKey和message兩個參數：
（1）routingKey為要發(fā)送的路由地址。
（2）message為具體的消息內容。發(fā)送者和接收者的queuename必須一致，不然無法接收。
最后，測試驗證。
創(chuàng)建對應的測試類ApplicationTests，驗證消息發(fā)送和接收是否成功。
- @RunWith(SpringRunner.class)
- @SpringBootTest
- public class ApplicationTests {
- @Autowired
- Producer producer;
- @Test
- public void contextLoads() throws InterruptedException {
- producer.produce();
- Thread.sleep(1*1000);
- }
- }
在上面的示例中，首先注入生產者對象，然后調用produce()方法來發(fā)送消息。
最后，單擊Run Test或在方法上右擊，選擇Run 'contextLoads()'，運行單元測試程序，查看后臺輸出情況，結果如下圖所示。
通過上面的程序輸出日志可以看到，消費者已經收到了生產者發(fā)送的消息并進行了處理。這是常用的簡單使用示例。
二、發(fā)送和接收實體對象
Spring Boot支持對象的發(fā)送和接收，且不需要額外的配置。下面通過一個例子來演示RabbitMQ發(fā)送和接收實體對象。
1. 定義實體類
首先，定義發(fā)送與接收的對象實體User類，代碼如下：
- public class User implements Serializable {
- public String name;
- public String password;
- // 省略get和set方法
- }
在上面的示例中，定義了普通的User實體對象。需要注意的是，實體類對象必須繼承Serializable序列化接口，否則會報數據無法序列化的錯誤。
2. 定義消費者
修改Consumer類，將參數換成User對象。示例代碼如下：
- @Component
- public class Consumer {
- @RabbitHandler
- @RabbitListener(queuesToDeclare = @Queue("rabbitmq_queue_object"))
- public void process(User user) {
- System.out.println("消費者消費消息111user=====name：" + user.getName()+",password:"+user.getPassword());
- }
- }
其實，消費者類和消息處理方法和之前的類似，只不過將參數換成了實體對象，監(jiān)聽rabbitmq_queue_object隊列。
3. 定義生產者
修改Producer類，定義User實體對象，并通過convertAndSend方法發(fā)送對象消息。示例代碼如下：
- @Component
- public class Producer {
- @Autowired
- private RabbitTemplate rabbitTemplate;
- public void produce() {
- User user=new User();
- user.setName("weiz");
- user.setPassword("123456");
- System.out.println("生產者生產消息111=====" + user);
- rabbitTemplate.convertAndSend("rabbitmq_queue_object", user);
- }
- }
在上面的示例中，還是調用convertAndSend()方法發(fā)送實體對象。convertAndSend()方法支持String、Integer、Object等基礎的數據類型。
4. 驗證測試
創(chuàng)建單元測試類，注入生產者對象，然后調用produceObj()方法發(fā)送實體對象消息，從而驗證消息能否被成功接收。
- @RunWith(SpringRunner.class)
- @SpringBootTest
- public class ApplicationTests {
- @Autowired
- Producer producer;
- @Test
- public void testProduceObj() throws InterruptedException {
- producer.produceObj();
- Thread.sleep(1*1000);
- }
- }
最后，單擊Run Test或在方法上右擊，選擇Run 'contextLoads()'，運行單元測試程序，查看后臺輸出情況，運行結果如圖12-8所示。
通過上面的示例成功實現了RabbitMQ發(fā)送和接收實體對象，使得消息的數據結構更加清晰，也更加貼合面向對象的編程思想。
