【RabbitMQ、Spring Boot】Spring Boot整合RabbitMQ实现并发和限流_springboot rabbitmq 限流

一、并发：一个listener对应多个consumer

介绍：默认情况一下，一个listener对应一个consumer。

方式一：通过配置文件配置(全局配置)

spring:
  rabbitmq:
    listener:
      simple:
        # 最小
        concurrency: 5
        # 最大
        max-concurrency: 10
1
2
3
4
5
6
7
8

方式二：在@RabbitListener上配置，只针对当前消费者有效

@RabbitListener(queues = "work_queue", concurrency = "3-10")
1

二、限流：consumer单位时间内接收到消息数量

方式一：通过配置文件配置(全局配置)

spring:
  rabbitmq:
    listener:
      simple:
        # 预处理模式更改为每次读取1条消息,在消费者未回执确认之前,不在进行下一条消息的投送
        prefetch: 3
1
2
3
4
5
6

方式二：在@RabbitListener上配置，只针对当前消费者有效

@RabbitListener(queues = "work_queue", containerFactory = "mqCachingConnectionFactory")
1

定义监听器工厂类

package com.cyun.demo.rabbitmq.consumer.config;

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * 监听器工厂类
 *
 * @author He PanFu
 * @date 2022-04-18 09:18:47
 */
@Configuration
public class RabbitMqConfig {

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Bean(name = "mqCachingConnectionFactory")
    public SimpleRabbitListenerContainerFactory mqCachingConnectionFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(50);
        return factory;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

参考链接：springboot整合rabbitmq实现限流与并发