MQ消息丢失怎么解决？5种方案帮你彻底搞定

// rabbitmq生产者确认配置
@bean
public rabbittemplate rabbittemplate() {
    rabbittemplate template = new rabbittemplate(connectionfactory);
    template.setconfirmcallback((correlationdata, ack, cause) -> {
        if (ack) {
            // 消息成功到达broker
            messagestatusservice.markconfirmed(correlationdata.getid());
        } else {
            // 发送失败，触发重试
            retryservice.scheduleretry(correlationdata.getid());
        }
    });
    return template;
}

// 可靠发送方法
public void sendreliable(string exchange, string routingkey, object message) {
    string messageid = generateid();
    // 先落库保存发送状态
    messagestatusservice.savesendingstatus(messageid, message);

    // 发送持久化消息
    rabbittemplate.convertandsend(exchange, routingkey, message, msg -> {
        msg.getmessageproperties().setdeliverymode(messagedeliverymode.persistent);
        msg.getmessageproperties().setmessageid(messageid);
        return msg;
    }, new correlationdata(messageid));
}

// 持久化队列配置
@bean
public queue orderqueue() {
    return queuebuilder.durable("order.queue")  // 队列持久化
            .deadletterexchange("order.dlx")    // 死信交换机
            .build();
}

// 发送持久化消息
public void sendpersistentmessage(object message) {
    rabbittemplate.convertandsend("order.exchange", "order.create", message, msg -> {
        msg.getmessageproperties().setdeliverymode(messagedeliverymode.persistent); // 消息持久化
        return msg;
    });
}

// kafka持久化配置
@bean
public producerfactory<string, object> producerfactory() {
    map<string, object> props = new hashmap<>();
    props.put(producerconfig.acks_config, "all"); // 所有副本确认
    props.put(producerconfig.retries_config, 3);   // 重试次数
    props.put(producerconfig.enable_idempotence_config, true); // 幂等性
    returnnew defaultkafkaproducerfactory<>(props);
}

    // 手动确认消费者
@rabbitlistener(queues = "order.queue")
public void handlemessage(order order, message message, channel channel) {
    long deliverytag = message.getmessageproperties().getdeliverytag();

    try {
        // 业务处理
        orderservice.processorder(order);

        // 手动确认
        channel.basicack(deliverytag, false);
        log.info("消息处理完成: {}", order.getorderid());

    } catch (exception e) {
        log.error("消息处理失败: {}", order.getorderid(), e);

        // 处理失败，重新入队
        channel.basicnack(deliverytag, false, true);
    }
}

// 消费者容器配置
@bean
public simplerabbitlistenercontainerfactory containerfactory() {
    simplerabbitlistenercontainerfactory factory = new simplerabbitlistenercontainerfactory();
    factory.setacknowledgemode(acknowledgemode.manual); // 手动确认
    factory.setprefetchcount(10); // 预取数量
    factory.setconcurrentconsumers(3); // 并发消费者
    return factory;
}

// 本地事务表方案
@transactional
public void createorder(order order) {
    // 1. 保存订单到数据库
    orderrepository.save(order);

    // 2. 保存消息到本地消息表
    localmessage localmessage = new localmessage();
    localmessage.setbusinessid(order.getorderid());
    localmessage.setcontent(json.tojsonstring(order));
    localmessage.setstatus(messagestatus.pending);
    localmessagerepository.save(localmessage);

    // 3. 事务提交，本地业务和消息存储保持一致性
}

// 定时任务扫描并发送消息
@scheduled(fixeddelay = 5000)
public void sendpendingmessages() {
    list<localmessage> pendingmessages = localmessagerepository.findbystatus(messagestatus.pending);

    for (localmessage message : pendingmessages) {
        try {
            // 发送消息到mq
            rabbittemplate.convertandsend("order.exchange", "order.create", message.getcontent());

            // 更新消息状态为已发送
            message.setstatus(messagestatus.sent);
            localmessagerepository.save(message);

        } catch (exception e) {
            log.error("发送消息失败: {}", message.getid(), e);
        }
    }
}

// rocketmq事务消息
public void sendtransactionmessage(order order) {
    transactionmqproducer producer = new transactionmqproducer("order_producer");

    // 发送事务消息
    message msg = new message("order_topic", "create", json.tojsonbytes(order));

    transactionsendresult result = producer.sendmessageintransaction(msg, null);

    if (result.getlocaltransactionstate() == localtransactionstate.commit_message) {
        log.info("事务消息提交成功");
    }
}

// 重试队列配置
@bean
public queue orderqueue() {
    return queuebuilder.durable("order.queue")
            .withargument("x-dead-letter-exchange", "order.dlx") // 死信交换机
            .withargument("x-dead-letter-routing-key", "order.dead")
            .withargument("x-message-ttl", 60000) // 60秒后进入死信
            .build();
}

// 死信队列配置
@bean
public queue orderdeadletterqueue() {
    return queuebuilder.durable("order.dead.queue").build();
}

// 消费者重试逻辑
@rabbitlistener(queues = "order.queue")
public void handlemessagewithretry(order order, message message, channel channel) {
    long deliverytag = message.getmessageproperties().getdeliverytag();

    try {
        orderservice.processorder(order);
        channel.basicack(deliverytag, false);

    } catch (temporaryexception e) {
        // 临时异常，重新入队重试
        channel.basicnack(deliverytag, false, true);

    } catch (permanentexception e) {
        // 永久异常，直接确认进入死信队列
        channel.basicack(deliverytag, false);
        log.error("消息进入死信队列: {}", order.getorderid(), e);
    }
}

// 死信队列消费者
@rabbitlistener(queues = "order.dead.queue")
public void handledeadlettermessage(order order) {
    log.warn("处理死信消息: {}", order.getorderid());
    // 发送告警、记录日志、人工处理等
    alertservice.sendalert("死信消息告警", order.tostring());
}