RabbitMQ实现消息的最终一致性

通过rabbit死信队列实现消息的最终一致性

配置消费队列

/**
 * @author gll
 * 2019/5/6 17:22
 */
@Configuration
public class RabbitConfiguration {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    protected void init() {
        //设置为支持事务
        rabbitTemplate.setChannelTransacted(true);
    }

    @Bean(name = "orderExchange")
    public Exchange rabbitExchange() {
        return new TopicExchange(DtsConstants.RABBIT_EXCHANGE, true, false);
    }

    @Bean(name = "orderQueue")
    public Queue orderQueue() {
        Map<String, Object> args = new HashMap<>(2);
        args.put("x-dead-letter-exchange", DtsConstants.RABBIT_DEADLETTER_EXCHANGE);
        args.put("x-dead-letter-routing-key", DtsConstants.RABBIT_DEADLETTER_ROUTINGKEY);
        return new Queue(DtsConstants.RABBIT_QUEUE, true, false, true, args);
    }

    @Bean(name="orderBinding")
    public Binding orderBinding(@Qualifier("orderQueue") Queue orderQueue,@Qualifier("orderExchange") Exchange orderExchange){
        return BindingBuilder.bind(orderQueue).to(orderExchange).with("order-add").noargs();
    }
}

配置死信队列

被拒绝的消息会进入死信队列


/**
 * @author gll
 * 2019/5/8 10:09
 */
@Configuration
public class RabbitDeadLetterConfiguration {

    /**
     * 订单死信队列
     *
     * @return
     */
    @Bean(name = "deadQueueLetter")
    public Queue deadLetterQueue() {
        return new Queue(DtsConstants.RABBIT_DEADLETTER_QUEUE, true, false, false, null);
    }

    @Bean(name = "deadLetterExchage")
    public Exchange deleteLetterExchange() {
        return new DirectExchange(DtsConstants.RABBIT_DEADLETTER_EXCHANGE, true, false);
    }

    @Bean(name = "deadLetterBinding")
    public Binding deadLetterBinding() {
        return new Binding(DtsConstants.RABBIT_DEADLETTER_QUEUE, Binding.DestinationType.QUEUE,
                DtsConstants.RABBIT_DEADLETTER_EXCHANGE, DtsConstants.RABBIT_DEADLETTER_ROUTINGKEY, null);
    }
}

正常数据消费者

/**
 * @author gll
 * 2019/5/6 17:25
 */
@Component
@RabbitListener(queues = {DtsConstants.RABBIT_QUEUE})
public class RabbitListenerTest {
    @RabbitHandler
    public void receive(String event) throws IOException {
        if(event.startsWith("error")){
            //消费者处理过程中，失败了，需要抛出异常。当超过重试次数的时候，消息进入死信队列
            System.out.println(event+"+"+new Date());
            throw new RuntimeException("error");
        }else{
            //消费者正常消费数据
            System.out.println(event);
        }
    }
}

死信队列消费者

/**
 * @author gll
 * 2019/5/6 17:25
 */
@Component
@RabbitListener(queues = {DtsConstants.RABBIT_DEADLETTER_QUEUE})
public class RabbitDeadLetterListener {
    @RabbitHandler
    @Transactional(rollbackFor = Exception.class)
    public void receive(String event) {
        //这里接收到异常消费的消息，可以做回滚，或者记录日志，以保证数据的最终一致性。
        System.out.println("dead-letter:"+event);
    }
}

事务消息

消息发送跟随事务一起提交，只有事务提交了，消息才会发送出去。

测试

TestController.java 包含三个方法

/rabbit/send:正常发送接口，生产者正常发送消息，消费者正常消费消息
/rabbit/send-error:异常发送接口，生产者发送消息后，出现异常，事务回滚，消息不会正常发送，消费者不消费
/rabbit/receive-error:消费者消费异常接口，生产者正常发送消息后，消费者消费过程中出现异常，经过重试次数之后，消息进入死信队列

完整代码：
https://github.com/qq54903099/spring-cloud-demo/tree/master/dead-letter

最后编辑于：2019.05.08 10:45:09