• 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、消费者代码

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

      ~本文完~

  • 相关阅读:
    vue3-element-plus使用
    python 进程 (概念+示例代码)
    C语言动态内存管理
    【JavaScript-32】使用js获取css样式内容
    mac电脑zsh: command not found: adb
    SpringCloud Alibaba Nacos配置中心快速搭建
    Ubuntu中安装mysql
    【C++11】initializer_list详解!
    基于SpringBoot+Vue网上商城系统的设计与实现
    Spring使用RestTemplate返回的嵌套实体对象为空,转换json报错
  • 原文地址:https://www.cnblogs.com/skevin/p/15956070.html