10. Fallback message

rabbitmq Fallback message

mandatory Parameters

When only the producer confirmation mechanism is turned on , After the switch receives the message , A confirmation message will be sent directly to the message producer , If the message is found to be non routable , Then the message will be discarded directly , At this time, the producer does not know that the message is discarded .

Can be set by mandatory Parameter returns the message to the producer when the destination cannot be reached during message delivery .

Case presentation

add to application.properties The configuration file

server.port=8080

spring.rabbitmq.host=172.16.140.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

spring.rabbitmq.publisher-confirm-type=correlated

Add configuration class

/** *  Fallback message configuration class  */
@Configuration
public class ConfirmConfig {
    

    public static final String CONFIRM_EXCHANGE = "confirm.exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";


    @Bean("confirmExchange")
    public DirectExchange confirmExchange(){
    
        return new DirectExchange(CONFIRM_EXCHANGE);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue(){
    
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    @Bean
    public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange){
    
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }

}

Add callback interface

/** *  Message fallback callback interface  */
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
    


    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    
        String id = correlationData != null ? correlationData.getId():"";
        if (ack){
    
            log.info(" The switch has received id by :{} The news of ",id);
        }else {
    
            log.info(" The switch hasn't received id by ：{} The news of , For some reason ：{}",id ,cause);
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
    
        log.info(" news ：{} Returned by server , Reasons for return ：{}, The switch is ：{}, route key：{}",new String(returned.getMessage().getBody()),returned.getReplyText(),returned.getExchange(),returned.getRoutingKey());
    }
}

Add consumer

/** *  Message consumer  */
@Component
@Slf4j
public class ConfirmConsumer {
    
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMsg(Message message){
    
        String msg = new String(message.getBody());
        log.info(" Received queue confirm.queue news ：{}",msg);
    }
}

Add producers

/** *  Message producer  */
@RestController
@RequestMapping("/confirm")
@Slf4j
public class ProducerController {
    
    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MyCallBack myCallBack;

    /** *  initialization rabbitTemplate */
    @PostConstruct
    public void init(){
    
        rabbitTemplate.setConfirmCallback(myCallBack);
        /** * true： When the switch cannot route messages , The message will be returned to the producer  * false： If a message is found to be unable to route , Throw away  */
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setReturnsCallback(myCallBack);
    }



    @GetMapping("sendMessage/{message}")
    public void sendMessage(@PathVariable String message){
    
        CorrelationData correlationData = new CorrelationData("1");
        String routingKey = "key1";

        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey,message+routingKey,correlationData);

        CorrelationData correlationData2 = new CorrelationData("2");
        String routingKey2 = "key2";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey2,message+routingKey,correlationData2);

        log.info(" Send message content ：{}",message);
    }
}

result