RabbitMq保证消息可靠性

缺乏、安全感 2022-12-09 01:58 549阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70][]

* 生产端消ConfirmCallback,ReturnCallback,消费端ACK机制
    * 1.做好消息确认机制(publisher,consumer(手动ACK))
    * 2.每一个发送的消息都在数据库做好记录,定期将失败的消息再次发送

try {
                    //TODO 确保每个消息发送成功，给每个消息做好日志记录，(给数据库保存每一个详细信息)保存每个消息的详细信息
                    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
                    rabbitTemplate.convertAndSend("order-event-exchange", "order.release.other", orderTo, correlationData);
                    MqMessage mqMessage = new MqMessage();
                    mqMessage.setMessageId(correlationData.getId());
                    mqMessage.setContent(JSON.toJSONString(orderTo));
                    mqMessage.setToExchange("order-event-exchange");
                    mqMessage.setRoutingKey("order.release.other");
                    mqMessage.setClassType("com.java.common.to.mq.OrderTo");
                    mqMessage.setMessageStatus(1);
                    mqMessage.setCreateTime(new Date());
                    save(mqMessage);
                } catch (Exception e) {
                    //TODO 定期扫描数据库，重新发送失败的消息
                }

CREATE TABLE `mq_message` (
      `message_id` varchar(32) NOT NULL COMMENT '消息ID',
      `content` text COMMENT '消息内容的json字符串',
      `to_exchange` varchar(255) DEFAULT NULL COMMENT '交换机',
      `routing_key` varchar(255) DEFAULT NULL COMMENT '路由键',
      `class_type` varchar(255) DEFAULT NULL COMMENT '实体类型',
      `message_status` int(1) DEFAULT '0' COMMENT '状态 0新建1已发送2错误抵达3已抵达',
      `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
      `update_time` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
      PRIMARY KEY (`message_id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

/**
         * 定制RabbitTemplate,确保消息不丢失
         * 生产端消ConfirmCallback,ReturnCallback
         * 消费端ACK机制
         */
        @PostConstruct
        public void initRabbitTemplate() {
            // ConfirmCallback消息抵达交换机的回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                /**
                 * 交换机(Exchange)收到消息就会回调
                 * CorrelationData 当前消息的唯一关联数据
                 * ack 消息是否成功送达
                 * cause 失败的原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    log.info("ID:[{}]的消息成功投递到交换机",correlationData.getId());
                    // 修改`mq_message`的`message_status`为1
                }
            });
            // ReturnCallback消息抵达队列的回调
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                /**
                 * 只要消息没有投递给指定的队列就会触发该回调
                 * @param message 投递失败的消息
                 * @param replayCode 回复的状态码
                 * @param replayText 回复的文本内容
                 * @param exchange 交换机
                 * @param routingKey 路由key
                 */
                @Override
                public void returnedMessage(Message message, int replayCode, String replayText, String exchange, String routingKey) {
                    log.error("ID:[{}]的消息失败投递到队列",message.getMessageProperties().getMessageId());
                    // 修改`mq_message`的`message_status`为2
                }
            });
        }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70]: /images/20221123/496a5e53f49d47baa4c17ccb78dc8ca1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 1]: /images/20221123/2885abffadb849feae776dc95cf71305.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 2]: /images/20221123/33921fb430984785a28676bcea11c039.png