Spring Boot整合RabbitMQ

• Spring Boot整合RabbitMQ
  • 寫在開頭
  • 整合流程
    • 配置檔案
    • RabbitMQ配置
    • RabbitMQ生產者
    • RabbitMQ消費者
      • 手動應答簡單工具類
  • 參考文件

寫在開頭

最近在搭一套基於SpringBoot的專案，用到了ssm+mysql+rabbitmq+redis。除了rabbitmq之外，其他幾個都很快整合好了，唯獨rabbitmq找了不少資料，才最終整合好，達到了預期。特此將過程記錄下來，供參考。

整合流程

整合流程中的程式碼都為整合的關鍵配置及其使用。至於SpringBoot的基本配置，請參考Spring Boot Quick Start

配置檔案

• pom.xml

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

• application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual # 手動應答
        concurrency: 5 # 消費端最小併發數
        max-concurrency: 10 # 消費端最大併發數
        prefetch: 5 # 一次請求中預處理的訊息數量
    cache:
      channel:
        size: 50 # 快取的channel數量
### 自定義配置
mq:
  defaultExchange: amqpExchange # 預設交換器
  queue: queue # 佇列名
  routeKey: queue_key # 路由key
 

• MQProperties

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Component
@ConfigurationProperties(prefix = "mq")
public class MQProperties {
    private String defaultExchange;
    private String routeKey;
    private String queue;

    public String getDefaultExchange() {
        return defaultExchange;
    }

    public void setDefaultExchange(String defaultExchange) {
        this.defaultExchange = defaultExchange;
    }

    public String getRouteKey() {
        return routeKey;
    }

    public void setRouteKey(String routeKey) {
        this.routeKey = routeKey;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }
}
 

RabbitMQ配置

import com.switchvov.rabbitmq.constant.MQProperties;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableRabbit
public class RabbitMQConfig {
    @Autowired
    private MQProperties mqProperties;

    @Bean
    public Queue queue() {
        boolean durable = true;
        boolean exclusive = false;
        boolean autoDelete = false;
        return new Queue(mqProperties.getQueue(), durable, exclusive, autoDelete);
    }

    @Bean
    public DirectExchange defaultExchange() {
        boolean durable = true;
        boolean autoDelete = false;
        return new DirectExchange(mqProperties.getDefaultExchange(), durable, autoDelete);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue())
                .to(defaultExchange())
                .with(mqProperties.getRouteKey());
    }
}

RabbitMQ生產者

import com.switchvov.rabbitmq.constant.MQProperties;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MQProperties mqProperties;

    @Test
    public void testSendMessage() {
        rabbitTemplate.convertAndSend(mqProperties.getDefaultExchange(), 
	        mqProperties.getRouteKey(), "傳送了一條資訊");
    }
}

RabbitMQ消費者

import com.switchvov.rabbitmq.common.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQService {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQService.class);

    @RabbitListener(queues = "${mq.queue}")
    public void receive(String payload, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        LOGGER.info("消費內容為：{}", payload);
        RabbitMQUtils.askMessage(channel, tag, LOGGER);
    }
}

手動應答簡單工具類

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;

import java.io.IOException;

public final class RabbitMQUtils {

    public static void askMessage(Channel channel, long tag, final Logger logger) {
        askMessage(channel, tag, logger, false);
    }

    public static void askMessage(Channel channel, long tag, final Logger logger, boolean multiple) {
        try {
            channel.basicAck(tag, multiple);
        } catch (IOException e) {
            logger.error("RabbitMQ，IO異常，異常原因為：{}", e.getMessage());
        }
    }

    public static void rejectMessage(Channel channel, long tag, final Logger logger) {
        rejectMessage(channel, tag, logger, false, false);
    }

    public static void rejectAndBackMQ(Channel channel, long tag, final Logger logger) {
        rejectMessage(channel, tag, logger, false, true);
    }

    public static void rejectMessage(Channel channel, long tag, final Logger logger, boolean multiple, boolean request) {
        try {
            channel.basicNack(tag, multiple, request);
        } catch (IOException e) {
            logger.error("RabbitMQ，IO異常，異常原因為：{}", e.getMessage());
        }
    }
}

參考文件

1. RabbitMQ訊息佇列（一）: Detailed Introduction 詳細介紹
   • 簡單的瞭解下RabbitMQ的系統架構和各種術語
2. Spring Boot中使用RabbitMQ
   • 簡單的Spring Boot和RabbitMQ整合資料
3. Spring boot整合RabbitMQ
   • 弄清楚了自定義配置queue、exchange、binding
4. Spring RabbitMQ - using manual channel acknowledgement on a service with @RabbitListener configuration
   • 接收者部分基本上就是按照這上面的回答實現的
5. spring boot / cloud (十九) 併發消費訊息,如何保證入庫的資料是最新的?
   • 併發消費配置

分享並記錄所學所見