Spring Boot整合RabbitMQ

时间：2025-02-13 17:48:37

Spring Boot整合RabbitMQ

@(MQ)[rabbitmq,springboot]

  • Spring Boot整合RabbitMQ
    • 写在开头
    • 整合流程
      • 配置文件
      • RabbitMQ配置
      • RabbitMQ生产者
      • RabbitMQ消费者
        • 手动应答简单工具类
    • 参考文档

写在开头

最近在搭一套基于SpringBoot的项目，用到了ssm+mysql+rabbitmq+redis。除了rabbitmq之外，其他几个都很快整合好了，唯独rabbitmq找了不少资料，才最终整合好，达到了预期。特此将过程记录下来，供参考。

整合流程

整合流程中的代码都为整合的关键配置及其使用。至于SpringBoot的基本配置，请参考Spring Boot Quick Start。

配置文件

<!-- rabbit-mq -->
<dependency>
    <groupId></groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
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 ;
import ;

@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 ;
import .*;
import ;
import ;
import ;
import ;

@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((), durable, exclusive, autoDelete);
    }

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

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

RabbitMQ生产者

import ;
import ;
import ;
import ;
import ;
import ;
import .;

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

    @Autowired
    private MQProperties mqProperties;

    @Test
    public void testSendMessage() {
        ((), 
            (), "发送了一条信息");
    }
}

RabbitMQ消费者

import ;
import ;
import org.;
import org.;
import ;
import ;
import ;
import ;

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

    @RabbitListener(queues = "${}")
    public void receive(String payload, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        ("消费内容为：{}", payload);
        (channel, tag, LOGGER);
    }
}
手动应答简单工具类
import ;
import org.;

import ;

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 {
            (tag, multiple);
        } catch (IOException e) {
            ("RabbitMQ，IO异常，异常原因为：{}", ());
        }
    }

    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 {
            (tag, multiple, request);
        } catch (IOException e) {
            ("RabbitMQ，IO异常，异常原因为：{}", ());
        }
    }
}

参考文档

  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 (十九) 并发消费消息,如何保证入库的数据是最新的?
    • 并发消费配置

相关文章