我有一个kafka制作人和消费者.kafka制作人正在发布一个主题，并且获取数据并完成一些处理 . kafka消费者正在阅读另一个主题，关于主题1的数据处理是否成功，即主题2是成功还是失败消息 . 现在我开始我的消费者，然后将数据发布到主题1.我想制作 生产环境 者和消费者同步，即一旦 生产环境 者发布数据，消费者应该读取该数据的成功或失败消息，然后 生产环境 者应该继续下一组数据 .
Apache Kafka和发布/订阅消息传递通常寻求通过使用流异步事件来解耦 生产环境 者和消费者 . 您所描述的更像是批处理作业或同步远程过程调用（RPC），其中 生产环境 者和消费者显式耦合在一起 . 标准的Apache Kafka 生产环境 者/消费者API不支持此消息交换模式，但您始终可以在使用相关ID，消耗ACK和请求/响应消息的Kafka API之上编写您自己的简单包装器，以使您自己的界面具有行为如你所愿 .
我刚刚开始构建基于kafka的处理环境 .
我想模拟某些请求的同步调用 - 我明确地不想要实际的同步调用（否则我不会考虑kafka） .
我的想法是以下......
请求到达多线程入口点处理程序（可能是Web服务器）
处理程序（在接收线程上）将请求发送到kafka总线进行处理 - 它不等待响应 .
处理程序向同步模拟器（SS）注册收到未来作为回报
处理程序轮询未来，直到将来指示完成 . 这有效地阻止处理程序线程，直到超时或完成，并且需要模拟同步调用 .
与此同时......请求由总线上的消费者在某处处理消费者向总线发送完成消息
完成消息到达正在由同步模拟器（SS）读取的完成消费者 . SS将收到的完成消息与已注册的等待请求相匹配 . 在找到匹配时，SS将请求标记为完成 . 与请求关联的未来会观察更新并指示完成 . 轮询未来的线程恢复处理 . 结果将返回给调用者
在我实际执行上述内容之前，我不知道任何细节，但似乎可行 .
我感谢任何人提供的建设性意见！
2 回答
Apache Kafka和发布/订阅消息传递通常寻求通过使用流异步事件来解耦 生产环境 者和消费者 . 您所描述的更像是批处理作业或同步远程过程调用（RPC），其中 生产环境 者和消费者显式耦合在一起 . 标准的Apache Kafka 生产环境 者/消费者API不支持此消息交换模式，但您始终可以在使用相关ID，消耗ACK和请求/响应消息的Kafka API之上编写您自己的简单包装器，以使您自己的界面具有行为如你所愿 .
我刚刚开始构建基于kafka的处理环境 .
我想模拟某些请求的同步调用 - 我明确地不想要实际的同步调用（否则我不会考虑kafka） .
我的想法是以下......
请求到达多线程入口点处理程序（可能是Web服务器）
处理程序（在接收线程上）将请求发送到kafka总线进行处理 - 它不等待响应 .
处理程序向同步模拟器（SS）注册
收到未来作为回报
处理程序轮询未来，直到将来指示完成 . 这有效地阻止处理程序线程，直到超时或完成，并且需要模拟同步调用 .
与此同时......请求由总线上的消费者在某处处理消费者向总线发送完成消息
完成消息到达正在由同步模拟器（SS）读取的完成消费者 . SS将收到的完成消息与已注册的等待请求相匹配 . 在找到匹配时，SS将请求标记为完成 . 与请求关联的未来会观察更新并指示完成 . 轮询未来的线程恢复处理 . 结果将返回给调用者
在我实际执行上述内容之前，我不知道任何细节，但似乎可行 .
我感谢任何人提供的建设性意见！