在 Kafka 中重播消息答案

【问题标题】：Replay messages in Kafka在 Kafka 中重播消息
【发布时间】：2018-10-01 06:49:58
【问题描述】：

我在 Spring Boot 中使用 kafka，我正在尝试添加一个功能，允许我们启动服务并让它重播消息到某个时间。

消费者是这样设置的

public interface ProductScenarioStream {
    String SERVICE_REQUESTS_PRODUCT_PRICE = "serviceRequestsProductPrice";
    String SERVICE_CONCLUDES_PRODUCT_SCENARIO = "serviceConcludesProductScenario";

    @Output(SERVICE_REQUESTS_PRODUCT_PRICE)
    MessageChannel serviceRequestsProductPrice();

    @Input(SERVICE_CONCLUDES_PRODUCT_SCENARIO)
    SubscribableChannel serviceConcludesProductScenario();
}

和

@Service
@EnableBinding(ProductScenarioStream.class)
@Profile("stream")
public class ProductStreamServiceImpl implements ProductStreamService 
{
    @Resource
    private ProductScenarioStream productScenarioStream;

    @Override
    public void send(final ServiceRequestsProductPrice event) {
     ...
    }
 }

你知道我在哪里可以找到允许我在这种情况下倒回流上的偏移量的设置吗？

【问题讨论】：

标签： java apache-kafka spring-cloud-stream spring-kafka

【解决方案1】：

我假设您的意思是 replay 而不是 reply - 我已经编辑了您的问题。

Spring Cloud Stream 目前没有公开寻找偏移量的机制。

你可以改用spring-kafka的@KafkaListener；实现ConsumerSeekAware，它为您提供了在启动期间（或任何时间）寻找的机制。

【讨论】：