带有 kafka 消费者记录的并行流

Question

我有卡夫卡记录：

ConsumerRecords<String, Events> records = kafkaConsumer.poll(POLL_TIMEOUT);

我想使用并行流而不是多线程来运行以下代码。

                records.forEach((record) -> {
                Event event = record.value();

                       HTTPSend.send(event);

            });

我尝试了多线程，但我想尝试并行流：

for (ConsumerRecord<String, Event> record : records) {
                        executor.execute(new Runnable() {
                            @Override
                            public void run() {

                                        HTTPSend.send(Event);

                            }
                        });

                    }

实际上，我在 HTTP.send 中遇到了多线程问题（即使是 1 个线程的线程池）。我来了

"Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target"。 这是通过 https 的请求。此错误仅在第一次发出请求时出现。之后，异常消失。噗！

对于我正在使用的多线程：

int threadCOunt=1;
                BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(threadCOunt, true);
                RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
                ExecutorService executor = new ThreadPoolExecutor(threadCOunt, threadCOunt, 0L, TimeUnit.MILLISECONDS, queue, handler);

HTTPSend.send() 是：

long sizeSend = 0;
    SSLContext sc = null;

    try {
        sc = SSLContext.getInstance("TLS");
        sc.init(null, TRUST_ALL_CERTS, new SecureRandom());
    } catch (NoSuchAlgorithmException | KeyManagementException e) {
        LOGGER.error("Failed to create SSL context", e);
    }

    // Ignore differences between given hostname and certificate hostname
    HostnameVerifier hv = (hostname, session) -> true;

    // Create the REST client and configure it to connect meta
    Client client = ClientBuilder.newBuilder()
            .hostnameVerifier(hv)
            .sslContext(sc).build();

    WebTarget baseTarget = client.target(getURL()).path(HTTP_PATH);
    Response jsonResponse = null;

    try {
        StringBuilder eventsBatchString = new StringBuilder();
        eventsBatchString.append(this.getEvent(event));
        Entity<String> entity = Entity.entity(eventsBatchString.toString(), MediaType.APPLICATION_JSON_TYPE);
        builder = baseTarget.request();
        LOGGER.debug("about to send the event {} and URL {}", entity, getURL());
        jsonResponse = builder.header(HTTP_ACK_CHANNEL, guid.toString())
                .header("Content-type", MediaType.APPLICATION_JSON)
                .header("Authorization", String.format("Meta %s", eventsModuleConfig.getSecretKey()))
                .post(entity);

Answer 1

我知道你想做什么，但我不确定这是不是最好的主意（我也不确定它是否不是）。

Kafka 的poll / commit 模型允许简单的背压并在您崩溃时保留最后处理的项目。通过“立即”返回您的轮询循环，您是在告诉 Kafka“我已经准备好接受更多”，并且提交偏移量（手动或自动）告诉 Kafka 您已经成功读取到该点。

您似乎想要做的是尽可能快地读取 Kafka，提交偏移量，然后将 Kafka 记录放入执行程序队列，然后平衡每秒的请求数等。

我不能 100% 确定这是个好主意：如果您的应用崩溃了怎么办？您可能已经提交了一些实际上没有进入上游的 Kafka 消息。如果您确实想这样做，我建议您在完成Runnable 后手动提交偏移量（通过commitSync），而不是让高级消费者为您完成。

为什么要使用线程执行器：我认为这些也可以使用 Kafka 来完成。

您可能希望同时向网络服务器发布多条消息。一个分区良好的 Kafka 主题将允许多个消费者/消费者组消费者多个分区，因此 - 假设一个完美扩展的 HTTP 服务器 - 可以让您并行地将消息发布到您的服务器。支持基于进程的并发！

也许网络服务器不是完全可扩展的，或者这个请求很慢（比如每个请求需要 1 秒）：你需要限制网络服务器每秒接受的请求数，如果你有一个队列，你可能有一个几个线程在不备份 Kafka 的情况下发布。

在这种情况下，您可以将 max.poll.records 设置为您的 Web 服务器所需的可扩展值。可能还有更好的方法可以做到这一点，尽管它现在正在逃避我。

如果您的网络服务器需要很长时间才能响应，您可能会收到与心跳失败相关的错误。在这种情况下，我会将您转至this SO answer on the timeout / heartbeat topic。

我不会使用线程执行器，从而使同步 HTTP 请求看起来是异步的，而是使用像 Netty 这样的事件 HTTP 客户端，从而在没有基于线程的并发的情况下实现并行。