RabbitMQ之消费者并发消费

为什么要引入消费者的并发消费？
当生产者的推送速度是远远超过消费者的能力的，可以提高消费者的消费速度。比如在java中我们可以启动多个 JVM 进程，实现多进程的并发消费，从而加速消费的速度，在mq中也可以通过设置配置。
@RabbitListener 注解中，有 concurrency 属性，它可以指定并发消费的线程数。
下面演示消费者并发消费实现
配置文件yml

spring:rabbitmq:#host为一般模式 若集群模式 将key换成addresses的形式host: 192.168.9.104port: 5672#账号密码自行替换username: adminpassword: adminlistener:# 选择的 ListenerContainer 的类型。默认为 simple 类型type: simplesimple:# 每个 @ListenerContainer 的并发消费的线程数concurrency: 2# 每个 @ListenerContainer 允许的并发消费的线程数max-concurrency: 10

rabbitmq.listener.type 的枚举值可以参考 ContainerType

SIMPLE 对应 SimpleMessageListenerContainer 消息监听器容器。它一共有两类线程：

Consumer 线程，负责从 RabbitMQ Broker 获取 Queue 中的消息，存储到内存中的 BlockingQueue 阻塞队列中。
Listener 线程，负责从内存中的 BlockingQueue 获取消息，进行消费逻辑

==================================》配置类代码
@Configuration
public class DirectExchangeConfiguration {/*** 创建一个 Queue** @return Queue*/@Beanpublic Queue queue08() {// Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除return new Queue(Message08.QUEUE,true,false,false);}/*** 创建 Direct Exchange** @return DirectExchange*/@Beanpublic DirectExchange exchange08() {// name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它return new DirectExchange(Message08.EXCHANGE,true,false);}/*** 创建 Binding* Exchange：Message08.EXCHANGE* Routing key：Message08.ROUTING_KEY* Queue：Message08.QUEUE** @return Binding*/@Beanpublic Binding binding08() {return BindingBuilder.bind(queue08()).to(exchange08()).with(Message08.ROUTING_KEY);}
========================》生产者代码
@Component
public class Producer08 {@Resourceprivate RabbitTemplate rabbitTemplate;public void syncSend(String id, String routingKey) {// 创建 Message08 消息Message08 message = new Message08();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Message08.EXCHANGE, routingKey, message);}
}
===========================》消费者代码
@Component
// 开启并发消费
@RabbitListener(queues = Message08.QUEUE, concurrency = "2")
@Slf4j
public class Consumer08 {@RabbitHandlerpublic void onMessage(Message08 message) throws InterruptedException {log.info("[{}][Consumer08 onMessage][线程编号:{} 消息内容：{}]", LocalDateTime.now(), Thread.currentThread().getId(), message);// 模拟消费耗时，为了让并发消费效果更好的展示TimeUnit.SECONDS.sleep(1);}
}
@ResourceProducer08 producer08;@Testvoid mock() throws InterruptedException {TimeUnit.SECONDS.sleep(20);}@SneakyThrows@Testvoid syncSend() {// 循环发送十个，观察消费者情况for (int i = 0; i < 10; i++) {String id = UUID.randomUUID().toString();producer08.syncSend(id, Message08.ROUTING_KEY);}log.info("[{}][test producer08 syncSend] 发送成功", LocalDateTime.now());// 这里多睡一会，确保消息全部消费完成TimeUnit.SECONDS.sleep(10);}