基于Kafka的微服务：如何确认消息处理成功并合并多服务结果

背景 #

现有一个消息主题，由NLP服务和计算机视觉（CV）服务共同消费，原始消息格式如下：

{
    "id": 1234,
    "text": "I love pizza",
    "photo": "https://photo.service/photo001"
}

• NLP服务处理后会向topic 1生成消息：

{
    "id": 1234,
    "text": "I love pizza",
    "nlp": "pizza",
    "photo": "https://photo.service/photo001"
}

• CV服务处理后会向topic 2生成消息：

{
    "id": 1234,
    "text": "I love pizza",
    "photo": "https://photo.service/photo001",
    "cv": ["pizza", "restaurant", "cup", "spoon", "folk"]
}

当前存在两个核心问题：

1. 如何确认NLP服务和CV服务已成功处理消息？
2. 最终服务如何获取id为1234的消息在两个主题中的对应处理结果？（注：NLP与CV服务处理时长存在差异）

问题1：确认服务处理完成的可行方案 #

方案1：状态通知主题机制 #

• 让NLP/CV服务在成功处理消息并发送结果到目标主题后，额外向一个独立的「状态通知主题」发送处理完成消息，格式示例：

{
    "msg_id": 1234,
    "service_type": "nlp",
    "status": "success",
    "timestamp": 1699999999
}

• 配套失败重试逻辑：若结果消息发送失败，服务需自动重试，达到重试上限后发送失败状态通知，便于排查。

方案2：消息队列ACK确认机制 #

如果使用的消息队列（如Kafka、RabbitMQ）支持消费确认：

• NLP/CV服务需在成功处理并发送结果消息后，再向原始消息主题发送ACK，标记该消息已处理完成。
• 若处理失败，可发送NACK将消息重新入队，或直接转入死信队列归档，避免重复无效处理。

方案3：结果消息被动校验 #

最终服务监听topic 1和topic 2，当收到对应id的结果消息时，即可判定对应服务已处理完成。但这种方式无法区分「服务未处理」和「处理完成但结果未送达队列」的情况，仅适合对确认精度要求较低的场景。

问题2：最终服务获取聚合结果的可行方案 #

方案1：本地缓存+超时等待 #

• 最终服务同时监听topic 1和topic 2，收到消息后将结果存入缓存（如Redis），以msg_id为键，存储结构示例：

{
    "1234": {
        "nlp_result": "pizza",
        "cv_result": null,
        "create_time": 1699999990
    }
}

• 根据两个服务的最大处理时长设置超时阈值，当缓存中某msg_id的NLP、CV结果均非空，或触发超时后，执行后续业务逻辑。
• 超时场景可选择：放弃处理、触发结果重试查询，或标记为异常待人工介入。

方案2：协调者模式（强一致性场景） #

• 引入独立协调者角色，由协调者向NLP和CV服务发送处理指令，等待两者返回处理完成信号后，再通知最终服务拉取结果。
• 若某一服务超时或失败，协调者触发重试或补偿逻辑，确保最终能获取有效结果，适合对一致性要求高的场景。

方案3：消息属性查询 #

如果消息队列支持按属性查询（如Kafka的消费者组回溯查询）：

• 最终服务按需发起查询，分别向topic 1和topic 2请求拉取id=1234的对应消息。
• 该方案适合非实时、按需获取结果的场景。

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