我应该如何在 Kafka 中实现延迟主题架构

Question

如果我的消费者暂时无法处理消息，我希望它推送到延迟 5 分钟的主题，如果它无法从那里处理，我也希望它推送到延迟 30 分钟的主题。如果这里也失败了，想把它推送到死信队列。

5 分钟延迟主题：消费者应在第一次处理后 5 分钟后收听。

30 分钟延迟主题：消费者应在上次失败后 30 分钟后收听。

我应该如何设计延迟队列？失败后推送到 Kafka 主题很容易，但我的消费者/听众应该在 5 分钟或 30 分钟延迟后如何收听？

我正在使用 SpringKafka 让消费者从下面的主题中收听-

@KafkaListener(topics = "${kafka.topic}")
public void receive1(String payload) {
    logger.info("Getting message on receiver-1");
    submitPayloadToExecutor(payload);
}

我自己的项目有以下实现，有人可以指出后备和任何新建议吗？？

    @KafkaListener(topics = "${kafka.topic}")
    public void receive3(String payload) {

        logger.info("Getting message on receiver-3");
        submitPayloadToExecutor(payload);
    }

    private void submitPayloadToExecutor(String payload) {

        StartupService startupService = StartupServiceSingleton.INSTANCE.getStartupServiceInstance();

        ObjectMapper mapper = startupService.getConverter().getObjectMapper();

        PublishPostProcessorEntity publishPostProcessorEntity = null;
        try {
            publishPostProcessorEntity = mapper.readValue(payload, PublishPostProcessorEntity.class);
            
            sleepForDelayedPublishedEntity(publishPostProcessorEntity);

            // ... do some work

            topicExecutorService.submit(publishPostProcessorEntity);
        } catch (Exception e) {
            // Work on exception
        }
    }

    private void sleepForDelayedPublishedEntity(PublishPostProcessorEntity publishPostProcessorEntity) {

        if (publishPostProcessorEntity instanceof DelayedPublishPostProcessorEntity) {

            DelayedPublishPostProcessorEntity delayedPublishPostProcessorEntity = (DelayedPublishPostProcessorEntity) publishPostProcessorEntity;

            // Fetch the topicName and sleep based on the configuration
            long pushedTimeStamp = delayedPublishPostProcessorEntity.getPushedTimeStamp();
            delayedPublishPostProcessorEntity.setComingTopicName(delayedPublishPostProcessorEntity.getNextTopicName());

            long currentTimeStamp = System.currentTimeMillis();

            if (CMSKafkaConstants.FIVE_MINUTES_DELAYED_TOPIC
                    .equalsIgnoreCase(delayedPublishPostProcessorEntity.getNextTopicName())) {

                long timeElapsed = currentTimeStamp - pushedTimeStamp;
                if ((Long.parseLong(firstDelay)-timeElapsed) > 0) {
                    // wait for timeToWait
                    try {
                        Thread.sleep(Long.parseLong(firstDelay)-timeElapsed);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            } else if (CMSKafkaConstants.THRITY_MINUTES_DELAYED_TOPIC
                    .equalsIgnoreCase(delayedPublishPostProcessorEntity.getNextTopicName())) {

                long timeElapsed = currentTimeStamp - pushedTimeStamp;
                if ((Long.parseLong(secondDelay)-timeElapsed) > 0) {
                    try {
                        Thread.sleep(Long.parseLong(secondDelay)-timeElapsed);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            }

        }

    }

Answer 1

您可以根据需要通过KafkaListenerEndpointRegistry 停止和启动消费者（使用getListenerContainer(id).start()）。