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配置」中所聲明的匿名隊列實例。

  ~本文完~

Tags: