7le / shine-mq Goto Github PK

可靠消息的发送

消息的可靠发送可以认为是尽最大努力发送消息通知，有两种实现方法：

• 发送消息之前，把消息持久到数据库，状态标记为待发送，然后发送消息，如果发送成功，将消息改为发送成功。若有失败，定时任务将会定时从数据库捞取一定时间内未发送的消息，将消息发送。
• 实现方式与第一种类似，不同的是持久消息的数据库是独立的，并不耦合在业务系统中。发送消息之前，先发送一个预消息给某一个第三方的消息管理器，消息管理器将其持久到数据库，并标记状态为待发送，发送成功后，标记消息为发送成功。定时任务定时从数据库捞取一定时间内未发送的消息，回查业务系统是否要继续发送，根据查询结果来确定消息的状态。

在shine-mq已经做了两种实现方法共性的消息持久化工作，默认实现的持久消息数据库是redis，也可以自己实现Coordinator接口（自行实现的需要注入到spring容器）。而捞起的超时异常消息，要怎么处理就需要根据业务自行选择是否要回查。

另外这个定时任务是需要自己实现的，具体如何实现可以自由发挥了（例如可以在服务A里实现定时任务，或者有统一的daemon来做处理）。

• 分布式事务：消息可靠发送

如果redis挂了，持久化会丢最多1秒的数据，那么这部分异常的消息怎么处理？

关于死信队列

在shine-mq使用分布式事务的时候会默认创建死信队列，死信队列的配置如下：

public static final String DEAD_LETTER_QUEUE = "dead_letter_queue";

public static final String DEAD_LETTER_EXCHANGE = "dead_letter_exchange";

public static final String DEAD_LETTER_ROUTEKEY = "dead_letter_routekey";

当服务B接收mq的消息失败次数大于receiveMaxRetries（默认为3，可以配置），就会将失败的消息投递到死信队列。死信队列的消息同样是可以进行消费，而如何来处理就是各自按着业务来实现了。

    @Autowired
    RabbitmqFactory factory;

    @PostConstruct
    public void test() {
        //配置死信队列 失败时候处理
        factory.add(MqConstant.DEAD_LETTER_QUEUE, MqConstant.DEAD_LETTER_EXCHANGE,
                MqConstant.DEAD_LETTER_ROUTEKEY, new ProcessorException(), SendTypeEnum.DLX);
    }

    /**
     * 处理异常消息
     */
    static class ProcessorException extends BaseProcessor {

        @Override
        public Object process(Object msg, Message message, Channel channel) {
            System.out.println("自行实现 通知人工处理 或者回调原服务A的回滚接口：" + msg);
            return null;
        }
    }

因为就实现了一条死信队列，多个分布式事务失败的消息都会被同一个死信队列的processor消费，这里可以根据Message获取对应原始的队列信息来区分处理。

发送ready消息的时候，ready消息发送成功，但是此时因为特殊原因，调用方业务回滚了（比如说宕机），那么ready消息被下游服务消费了，但是实际上上游服务是没成功的，不就有问题了吗？