精通SpringBoot---整合RabbitMQ消息隊列

今天來和朋友們一塊兒學習下，SpringBoot怎麼整合RabbitMQ。目前消息組件大體有三種：.activemq, rabbitmq, kafka。這三者各有優缺點，RabbitMQ相比之下是處於其餘兩者之間的一個消息組件。RabbitMQ依賴於erlang，在linux下安裝的話，要先安裝erlang環境。下面來看看怎麼SpringBoot 怎麼整合RabbitMQ吧。

1. 想要使用RabbitMQ ，pom依賴是少不了的~

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.再來看看application.yml文件的內容

spring:
  rabbitmq:
    username: rabbit
    password: 123456 host: localhost port: 5672 virtual-host: / #手動ACK 不開啓自動ACK模式,目的是防止報錯後未正確處理消息丟失 默認 爲 none listener: simple: acknowledge-mode: manual

RabbitMQConfig的內容（註冊）

import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { public static final String DEFAULT_MAIL_QUEUE = "dev.mail.register.default.queue"; public static final String MANUAL_MAIL_QUEUE = "dev.mail.register.manual.queue"; @Bean public Queue defaultMailQueue (){ // Queue queue = new Queue(Queue名稱，消息是否須要持久化處理) return new Queue(DEFAULT_MAIL_QUEUE, true); } @Bean public Queue manualMailQueue(){ return new Queue(MANUAL_MAIL_QUEUE, true); } }

搞兩個監聽器（使用@RabbitListener註解）來監聽下這兩種消息 (怎麼感受本身如今說話一股土味兒，最近吃土吃多了麼~ 好吧，寫的代碼估計也是土味的吧）（監聽隊列）

import com.developlee.rabbitmq.config.RabbitMQConfig; import com.developlee.rabbitmq.entity.MailEntity; import com.rabbitmq.client.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; @Component public class MailHandler { private static final Logger logger = LoggerFactory.getLogger(MailHandler.class); /** * <p>TODO 該方案是 spring-boot-data-amqp 默認的方式,不太推薦。具體推薦使用 listenerManualAck()</p> * 默認狀況下,若是沒有配置手動ACK, 那麼Spring Data AMQP 會在消息消費完畢後自動幫咱們去ACK * 存在問題：若是報錯了,消息不會丟失,可是會無限循環消費,一直報錯,若是開啓了錯誤日誌很容易將磁盤空間耗完 * 解決方案：手動ACK,或者try-catch 而後在 catch 裏面講錯誤的消息轉移到其它的系列中去 * spring.rabbitmq.listener.simple.acknowledge-mode=manual * <p> * * @param mail 監聽的內容 */ @RabbitListener(queues = {RabbitMQConfig.DEFAULT_MAIL_QUEUE}) public void listenerAutoAck(MailEntity mail, Message message, Channel channel) { //TODO 若是手動ACK,消息會被監聽消費,可是消息在隊列中依舊存在,若是 未配置 acknowledge-mode 默認是會在消費完畢後自動ACK掉 final long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { logger.info("listenerAutoAck 監聽的消息-{}", mail.toString()); //TODO 通知MQ 消息已被成功消費，能夠ACK了 channel.basicAck(deliveryTag, false); } catch (IOException e) { //處理失敗， 從新壓入MQ. try { channel.basicRecover(); } catch (IOException e1) { e1.printStackTrace(); } } } @RabbitListener(queues = {RabbitMQConfig.MANUAL_MAIL_QUEUE}) public void listenerManualAck(MailEntity mail, Message message, Channel channel) { logger.info("listenerManualAck 監聽的消息-{}", mail.toString()); try { //TODO 通知MQ 消息已被成功消費，能夠ACK了 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { //若是報錯，容錯處理， } } }

再來一波測試代碼，測試下......（實例化隊列）

import com.developlee.rabbitmq.config.RabbitMQConfig; import com.developlee.rabbitmq.entity.MailEntity; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * @author Lee * @// TODO 2018/6/22-11:20 * @description */ @RestController @RequestMapping(value = "/mail") public class MailController { private final RabbitTemplate rabbitTemplate; @Autowired public MailController(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; } /** * this.rabbitTemplate.convertAndSend(RabbitConfig.DEFAULT_MAIL_QUEUE, mailEntity); * 對應 {@link MailHandler#listenerAutoAck}; * this.rabbitTemplate.convertAndSend(RabbitConfig.MANUAL_MAIL_QUEUE, mailEntity); * 對應 {@link MailHandler#listenerManualAck}; */ @GetMapping("/default") public void defaultMailMsg() { MailEntity mailEntity = new MailEntity(); mailEntity.setId("1"); mailEntity.setName("First Mail Message"); mailEntity.setTitle("RabbitMQ with Spring boot!"); mailEntity.setContent("Come on! Let's study Micro-Service together!"); this.rabbitTemplate.convertAndSend(RabbitMQConfig.DEFAULT_MAIL_QUEUE, mailEntity); this.rabbitTemplate.convertAndSend(RabbitMQConfig.MANUAL_MAIL_QUEUE, mailEntity); } }

MailEntity.java

import java.io.Serializable; public class MailEntity implements Serializable { private static final long serialVersionUID = -2164058270260403154L; private String id; private String name; private String title; private String content; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getTitle() { return title; } public void setTitle(String title) { this.title = title; } public String getContent() { return content; } public void setContent(String content) { this.content = content; } }

啓動項目 ，瀏覽器地址欄輸入http://localhost:8080/mail。 something you will find in your heart。