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

博客 动态
0 213
羽尘
羽尘 2022-03-02 17:56:00
悬赏:0 积分 收藏

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

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

 

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

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

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

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

3、Exchange配置

    @Bean    public FanoutExchange myFanoutExchange() {       //“my.fanout”仅用于演示,请按自己的项目情况选择合适的、有意义的命名        return new FanoutExchange("my.fanout", true, false);    }        @Bean    public Queue anonymousQueue() {        //注意:必须使用匿名队列才能达到本文目标,此匿名队列实例的名称在消费端会被使用        return new AnonymousQueue();    }        @Bean    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, correlationData);        } catch (Exception e) {            logger.error(e.getMessage(), e);        }

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

5、消费者代码

@Servicepublic 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配置”中所声明的匿名队列实例。

  ~本文完~

posted @ 2022-03-02 17:12 SKevin 阅读(34) 评论(0) 编辑 收藏 举报
回帖
    羽尘

    羽尘 (王者 段位)

    2335 积分 (2)粉丝 (11)源码

     

    温馨提示

    亦奇源码

    最新会员