rabbitMQ对消息不可达处理-备份交换机/备份队列

生产者发送消息，在消息不可达指定队列时，可以借助扇出类型交换机（之前写过消息回退的处理方案，扇出交换机处理的方案优先级高于消息回退）处理不可达消息，然后放置一个备份队列，供消费者处理不可达消息，同时也加一个报警队列，对于不能走正常流程的消息进行消费者告警。

先用方法配置类把各个组件声明：

package com.esint.configs;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class BackupConfig {

    /**
     * 定义组件常量名字
     */
    //交换机- 确认交换机额
    public static final String EXCHANGE_SURE = "sure.ex";
    //交换机- 备份交换机额
    public static final String EXCHANGE_BACK = "backup.ex";

    //队列- 正常确认队列
    public static final String QUEUE_SURE = "sure.queue";
    //队列-备份队列
    public static final String QUEUE_BACKUP = "backup.queue";
    //队列-警告队列
    public static final String QUEUE_WARN = "warn.queue";

    //routing-key
    public static final String ROUTING_KEY_SURE = "key1";

    /**
     * 声明组件
     */
    //确认交换机
    @Bean("sureExchange")
    public DirectExchange sureExchange(){
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("alternate-exchange",EXCHANGE_BACK);
        return ExchangeBuilder.directExchange(EXCHANGE_SURE).durable(true).withArguments(arguments).build();

    }

    //备份交换机
    @Bean("backExchange")
    public FanoutExchange backExchange(){
        return new FanoutExchange(EXCHANGE_BACK);
    }

    //确认队列
    @Bean("sureQueue")
    public Queue sureQueue(){
        return QueueBuilder.durable(QUEUE_SURE).build();
    }

    //备份队列
    @Bean("backupQueue")
    public Queue backupQueue(){
        return QueueBuilder.durable(QUEUE_BACKUP).build();
    }

    //警告队列
    @Bean("warnQueue")
    public Queue warnQueue(){
        return QueueBuilder.durable(QUEUE_WARN).build();
    }

    /**
     * 绑定组件  确认队列 绑定 确认交换机 with key1
     */
    @Bean
    public Binding sureQueueBindingSureExchange(@Qualifier("sureQueue") Queue sureQueue,
                                             @Qualifier("sureExchange")DirectExchange sureExchange){
        return BindingBuilder.bind(sureQueue).to(sureExchange).with(ROUTING_KEY_SURE);

    }

    /**
     * 绑定组件 备份队列 绑定 备份交换机
     */
    @Bean
    public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
                                                    @Qualifier("backExchange")FanoutExchange backExchange){

        return  BindingBuilder.bind(backupQueue).to(backExchange);
    }

    /**
     * 绑定组件 警告队列 绑定 备份交换机
     */
    @Bean
    public Binding warnQueueBindingBackupExchange(@Qualifier("warnQueue") Queue warnQueue,
                                                  @Qualifier("backExchange")FanoutExchange backExchange){

        return  BindingBuilder.bind(warnQueue).to(backExchange);

    }
}

生产者： 我们做出两个方法，一个可正常进行流程，一个routingKey异常无法路由到指定队列

package com.esint.controller;

import com.esint.configs.BackupConfig;
import com.esint.constants.ResponseCode;
import com.esint.entity.ResponseEntity;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.apache.kafka.clients.producer.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.concurrent.ExecutionException;


@Api(value = "rabbitMQ-备份队列测试")
@RestController
@RequestMapping("/rabbit")
public class BackUpExchangeController {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @ApiOperation(value = "routingKey正常测试",httpMethod = "GET",tags = {"去正常流程"})
    @ApiImplicitParams({
            @ApiImplicitParam(name="str",value="消息体",required = false,dataType = "String")
    })
    @ResponseBody
    @RequestMapping(value = "/test1", method = RequestMethod.GET)
    public ResponseEntity test1(String str ) {

        rabbitTemplate.convertAndSend(BackupConfig.EXCHANGE_SURE, BackupConfig.ROUTING_KEY_SURE,str);
        return new ResponseEntity(ResponseCode.SUCCESS).addData("routingKeyOk:"+str);

    }

    @ApiOperation(value = "routingKey非正常测试",httpMethod = "GET",tags = {"去备份-警告"})
    @ApiImplicitParams({
            @ApiImplicitParam(name="str",value="消息体",required = false,dataType = "String")
    })
    @ResponseBody
    @RequestMapping(value = "/test2", method = RequestMethod.GET)
    public ResponseEntity test2(String str ) {

        rabbitTemplate.convertAndSend(BackupConfig.EXCHANGE_SURE, BackupConfig.ROUTING_KEY_SURE+"wrong",str);

        return new ResponseEntity(ResponseCode.SUCCESS).addData("routingKeyWrong:"+str);

    }

}

三个消费者分别监听正常队列 备份队列 警告队列

确认队列消费者：

package com.esint.consumer;

import com.esint.configs.BackupConfig;
import com.esint.configs.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class BackUpConsumer01 {

    @RabbitListener(queues = BackupConfig.QUEUE_SURE)
    public void reveiver(Message message){
        log.info("正常消费者C1：" +  new String(message.getBody()),"UTF-8");
    }

}

备份队列消费者：

package com.esint.consumer;

import com.esint.configs.BackupConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class BackUpConsumer02 {

    @RabbitListener(queues = BackupConfig.QUEUE_BACKUP)
    public void reveiver(Message message){
        log.info("备份消费者C2：" +  new String(message.getBody()),"UTF-8");
    }
}

警告队列消费者：

package com.esint.consumer;

import com.esint.configs.BackupConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class BackUpConsumer03 {

    @RabbitListener(queues = BackupConfig.QUEUE_WARN)
    public void reveiver(Message message){
        log.info("警告消费者C3：" +  new String(message.getBody()),"UTF-8");
    }

}

测试：

1.正常流程测试

 com.esint.consumer.BackUpConsumer01      : 正常消费者C1：你好啊 正常队列

2.路由不达消息测试

com.esint.consumer.BackUpConsumer03      : 警告消费者C3：这个消息不可达 routing-key不对 它去哪里了？
com.esint.consumer.BackUpConsumer02      : 备份消费者C2：这个消息不可达 routing-key不对 它去哪里了？

测试达到预期结果！

在队列消息不可达时，备份交换机处理优先级高于消息回退处理。

代码007普通

打赏 收藏 海报 链接