Pulsar send_async回调中ack消息后input_topic积压未减少的问题咨询

为什么消息积压没下降？ #

你的代码里用异步发送+回调ack的方式存在几个核心问题：

1. 回调执行延迟/丢失：send_async是异步操作，回调只有在发送成功/失败后才会触发。如果生产者发送缓慢、网络波动或者服务异常，回调可能迟迟不执行，甚至完全不执行，导致对应的消息永远无法被ack，积压持续增加。
2. Lambda变量捕获陷阱：如果是在循环中创建多个生产者和回调，lambda里的msg会被后续循环迭代覆盖，最终ack的可能是最后一条消息，前面的消息根本没被确认。
3. 异常处理缺失：如果send_async本身抛出异常（比如生产者初始化失败），回调不会执行，消息也得不到ack。

现有代码的修复方案 #

1. 修复Lambda变量捕获 #

把msg作为默认参数传入lambda，避免循环中变量被覆盖：

callback = lambda res, msg=msg: consumer.acknowledge(msg)
producers[i].send_async(msg, callback=callback)

2. 确保所有路径都能ack #

添加异常捕获，即使发送失败也要处理ack（根据业务需求选择直接ack或重试后ack）：

def handle_callback(res, msg, consumer):
    # 可在这里记录发送结果日志
    consumer.acknowledge(msg)

for msg in consumer:
    for i in range(len(producers)):
        try:
            producers[i].send_async(msg, callback=lambda res, msg=msg: handle_callback(res, msg, consumer))
        except Exception as e:
            # 发送失败时直接ack，或根据业务逻辑重试
            consumer.acknowledge(msg)

3. 调整生产者配置 #

检查生产者的acks参数：如果设置为all，需要等待所有副本确认才会触发回调，延迟较高。可以临时调整为1（等待主副本确认）来加快回调执行速度，权衡数据可靠性和ack效率。

最优替代方案 #

1. 使用Kafka Streams（推荐） #

这是Kafka官方提供的流处理框架，原生支持消息路由转发，自动管理offset提交，无需手动处理ack，可靠性和可维护性更高。示例代码（Python版）：

from confluent_kafka.streams import StreamsConfig, StreamsBuilder, Serdes

def main():
    config = {
        StreamsConfig.APPLICATION_ID_CONFIG: "forward-app",
        StreamsConfig.BOOTSTRAP_SERVERS_CONFIG: "kafka-broker:9092",
        StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG: Serdes.String().get_class(),
        StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG: Serdes.String().get_class()
    }

    builder = StreamsBuilder()
    input_stream = builder.stream("input_topic")

    # 方式1：按条件分支转发到不同主题
    branches = input_stream.branch(
        lambda key, value: "topic1" in value,
        lambda key, value: "topic2" in value
    )
    branches[0].to("topic1")
    branches[1].to("topic2")

    # 方式2：无条件转发到多个主题
    input_stream.foreach(lambda key, value: (
        Producer({"bootstrap.servers": "kafka-broker:9092"}).produce("topic1", key=key, value=value),
        Producer({"bootstrap.servers": "kafka-broker:9092"}).produce("topic2", key=key, value=value)
    ))

    streams = KafkaStreams(builder.build(), config)
    streams.start()

if __name__ == "__main__":
    main()

2. 改用同步发送（适合小流量场景） #

放弃异步发送，改用send_sync确保发送完成后立即ack，虽然吞吐量略低，但逻辑简单可靠：

for msg in consumer:
    success = True
    try:
        for producer in producers:
            producer.send_sync(msg)
    except Exception as e:
        success = False
        # 记录发送失败日志
    if success:
        consumer.acknowledge(msg)
    else:
        # 业务逻辑：重试N次后ack，或发送到死信队列
        consumer.acknowledge(msg)

3. 批量处理（高流量场景） #

积累一定数量的消息后批量发送，批量ack，提升吞吐量的同时减少ack操作的开销：

batch_size = 100
batch = []

for msg in consumer:
    batch.append(msg)
    if len(batch) >= batch_size:
        try:
            for producer in producers:
                producer.send_batch(batch)
            # 批量ack
            for msg in batch:
                consumer.acknowledge(msg)
            batch = []
        except Exception as e:
            # 处理批量失败，比如拆分重试或直接ack
            pass

内容的提问来源于stack exchange，提问作者user3863316