我有Kafka的记录：

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

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

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

                       HTTPSend.send(event);

            });

我尝试了mlutithreading，但我想尝试parallelstream：

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);