Spring Boot+RabbitMQ 通过fanout模式实现消息接收功能(支持消费者多实例部署)

Spring Boot+RabbitMQ 通过fanout模式实现消息接收功能(支持消费者多实例部署)

本文章适用的场景：同一条消息可以被多个消费者同时消费。注意：当消费者多实例部署时，会轮询消费消息。网上有大量的的案例展示：P生产一条消息，消费者服务C中建立Q1和Q2两个队列共同消费。但极少的材料展示：P生产一条消息后M1，消费者C1和C2可以同时消费M1，如下图所示。案例基于Spring Boot以及RabbitMQ的“fanout”类型exchange。已经实测可放心使用。

1、引入基本依赖，项目不同请您按自己的情况引入合适的依http://赖

org.springframework.boot

spring-boot-starter-amqp

2、rabbitmq密码、用户名等配置

内容过于基础，请网络上自行搜索相关内容。

3、Exchange配置

@Bean

public FanoutExchange myFanoutExchange() {

　 //“my.fanout”仅用于演示，请按自己的项目情况选择合适的、有意义的命名

return new FanoutExchange("my.fanout", true, false);

}

@Bean

public Queue anonymousQueue() {

//注意：必须使用匿名队列才能达到本文目标，此匿名队列实例的名称在消费端会被使用

return new AnonymousQueue();

public Binding bindAnonymousQueue() {

return BindingBuilder.bind(anonymousQueue()).to(myFanoutExchange());

4、发送端代码

//构建回调返回的数据

CorrelationData correlationData = new CorrelationData(UUIDUtils.getUUID());

Message msg = MessageBuilder.withBody(message.getBytes())

.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)

.setCorrelationId(correlationData.getId())

.build();

try {

rabbitTemplate.convertAndSend(exchangeName, null, msg, corrhttp://elationData);

} catch (Exception e) {

logger.error(e.getMessage(), e);

}

注意：“fanout"模式直接将消息发送到“exchange”，routing key应为null或“”，如果您的代码只需要发送消息而无接收，请忽略第3小节“Exchange配置”中所示代码

5、消费者代码

@Service

public class MyListener {

@RabbitListener(queues = "#{anonymousQueue.name}")

public void receive(String message, Message message1, Channel channel) {

try {

//your code

} catch (Exception e) {

logger.error(e.getMessage(), e);

} finally {

long deliverTag = message1.getMessageProperties().getDeliveryTag();

channel.basicAck(deliverTag, false);

}

}

}

代码“@RabbitListener…”中的“#{anonymousQueue.name}”中的“anonymousQueue”对应第3小节“Exchange配置”中所声明的匿名队列实例。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。