c++ - 如何在 librdkafka 中正确重新发送失败的消息？

Question

static void DeliverCallback(rd_kafka_t *rk,  void *payload, size_t len, rd_kafka_resp_err_t error_code,  void *opaque, void *msg_opaque)
{
    if (error_code != RD_KAFKA_RESP_ERR_NO_ERROR)
    {
            // todo resend the message

    }
}

我的dr_cb是这样的。我知道如何获取发送失败消息的内容，但是如何获取主题呢？最优雅的方法是什么？

Answer 1

dr_cb 接口(interface)已过时且已弃用，您应该使用更丰富的 dr_msg_cb 来代替，您可以在其中访问所有 rd_kafka_message_t 字段，对于您的情况，主题的 rkmessage->rkt 。

但是，您通常不应该在发送报告失败时重试生成消息，因为 librdkafka 将竭尽全力在配置的约束内生成消息(message.timeout.ms和重试)，应用程序通过再次重试相同的消息并不能增加多少值(value)。

而是配置 message.timeout.ms 来满足您的业务需求，回答“尝试生成这段数据需要多长时间才有意义？”的问题，并设置重试 到其最大值(因为从应用程序的角度来看，重试次数无关)。

如果重复或消息排序至关重要，您还应该考虑使用幂等生产者 (enable.idempotence=true)。

最后，librdkafka 为每个已发送或失败的消息提供了一个消息持久性指示器 API (rd_kafka_message_status())，让应用程序知道该消息是否...:

• 绝对不会坚持
• 可能持续存在(手动重试可能会导致重复)
• 绝对坚持

参见https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md#message-reliability有关 librdkafka 中消息可靠性的更多信息。