使用apache camel和Kafka时如何处理背压？

Question

我正试图编写一个应用程序，将与Kafka集成使用骆驼。（版本-3.4.2）

我从这个问题的答案中借用了一种方法。

我有一条路线可以监听Kafka主题的信息。通过使用一个简单的执行器，该消息的处理与消耗是分离的。每个处理都作为任务提交给该执行者。消息的顺序并不重要，唯一需要考虑的因素是消息处理的速度和效率。我已禁用自动提交，并在任务提交给执行者后手动提交消息。丢失当前正在处理的消息（由于崩溃/关闭）是可以的，但Kafka中从未提交处理的消息不应丢失（由于提交了偏移量）。现在来回答问题，

1. 如何有效地处理负载？例如，有1000条消息，但我一次只能并行处理100条

现在，我的解决方案是阻止消费者轮询线程，并尝试持续提交作业。但暂停投票会是一个更好的办法，但我找不到任何办法来实现这一点。

public static void main(String[] args) throws Exception {
        String consumerId = System.getProperty("consumerId", "1");
        ExecutorService executor = new ThreadPoolExecutor(100, 100, 0L, TimeUnit.MILLISECONDS,
                new SynchronousQueue<>());
        LOGGER.info("Consumer {} starting....", consumerId);

        Main main = new Main();
        main.init();

        CamelContext context = main.getCamelContext();
        ComponentsBuilderFactory.kafka().brokers("localhost:9092").metadataMaxAgeMs(120000).groupId("consumer")
                .autoOffsetReset("earliest").autoCommitEnable(false).allowManualCommit(true).maxPollRecords(100)
                .register(context, "kafka");

        ConsumerBean bean = new ConsumerBean();
        context.addRoutes(new RouteBuilder() {
            @Override
            public void configure() {
                from("kafka:test").process(exchange -> {
                    LOGGER.info("Consumer {} - Exhange is {}", consumerId, exchange.getIn().getHeaders());
                    processTask(exchange);
                    commitOffset(exchange);
                });
            }

            private void processTask(Exchange exchange) throws InterruptedException {
                try {
                    executor.submit(() -> bean.execute(exchange.getIn().getBody(String.class)));
                } catch (Exception e) {
                    LOGGER.error("Exception occured {}", e.getMessage());
                    Thread.sleep(1000);
                    processTask(exchange);
                }
            }

            private void commitOffset(Exchange exchange) {
                boolean lastOne = exchange.getIn().getHeader(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, Boolean.class);
                if (lastOne) {
                    KafkaManualCommit manual = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT,
                            KafkaManualCommit.class);
                    if (manual != null) {
                        LOGGER.info("manually committing the offset for batch");
                        manual.commitSync();
                    }
                } else {
                    LOGGER.info("NOT time to commit the offset yet");
                }
            }
        });

        main.run();
    }

Answer 1

为此，可以使用throttleEIP。

from("your uri here")
.throttle(maxRequestCount)
.timePeriodMillis(inTimePeriodMs)
.to(yourProcessorUri)
.end()

请看这里的原始留档。