将for循环转换为多线程块

Question

我在一个函数中有一个循环函数，我打算并行化但不确定多个线程的负载是否会超重并发的好处。

我只需要将不同的日志文件发送到相应的接收器。对于时间可以说接收器的数量不会超过10.而不是将日志文件背靠背发送，如果我将它们全部并行发送它会更有效吗？

for(int i=0; i < receiversList.size(); i++)
        {
            String receiverURL = serverURL + receiversList.get(i);
            HttpPost method = new HttpPost(receiverURL);

            String logPath = logFilesPath + logFilesList.get(i);
            messagesList = readMsg(logPath);

            for (String message : messagesList) {
                StringEntity entity = new StringEntity(message);
                log.info("Sending message:");
                log.info(message + "\n");
                method.setEntity(entity);
                if (receiverURL.startsWith("https")) {
                    processAuthentication(method, username, password);
                }
                httpClient.execute(method).getEntity().getContent().close();
            }
            Thread.sleep(500); // Waiting time for the message to be sent
        }

另外请告诉我如果能够工作的话我怎么能让它平行？我应该手动操作还是使用ExecutorService？

Answer 1

我只需要将不同的日志文件发送到相应的接收器。暂时可以说接收器的数量不会超过10.而不是将日志文件背靠背发送，如果我将它们全部并行发送会更有效吗？

在我们确定并行执行此操作之前会有很多问题要问你。你提到过“接收器”，但是你真的在谈论不同网址上的不同接收服务器，还是所有线程都将日志文件发送到同一台服务器？如果是后者，那么你很可能在并发性方面获得很小的改进。单个线程应该能够很好地填充网络管道。

此外，如果消息很小，您可能不会加速。只有大量的消息才会花费任何时间，并且如果它们是并行发送的话，可以给你任

我最熟悉ExecutorService课程。你可以这样做：

ExecutorService threadPool = Executors.newFixedThreadPool(10);
...
threadPool.submit(new Runnable() {
    // you could create your own Runnable class if each one needs its own httpClient
    public void run() {
          StringEntity entity = new StringEntity(message);
          ...
          // we assume that the client is some sort of pooling client
          httpClient.execute(method).getEntity().getContent().close();
      }
    }
});

如果您想将这些消息排队并在后台线程中发送它们以减慢您的程序速度，那将会有什么好处。然后你可以将消息提交给threadPool并继续前进。或者你可以把它们放在BlockingQueue<String>并从BlockingQueue获取一个线程并调用httpClient.execute(...)。

来自这个good ExecutorService tutorial的更多实现细节。

最后，如何将所有消息放入一个实体并在服务器上划分消息。尽管您可能无法控制服务器处理程序代码，但这将是最有效的。

Answer 2

Hello ExecutorService当然是一个选项。您有4种方法可以在Java中完成它。

1. 使用线程（暴露许多容易出错的细节）
2. 正如您已经提到的那样执行服务。它来自Java 6这是一个演示ExecutorService http://tutorials.jenkov.com/java-util-concurrent/executorservice.html的教程
3. ForkJoin框架来自Java 7
4. ParallelStreams来自Java 8，下面是使用ParallelStreams的解决方案

寻求更高级别的api将为您节省一些您可能会做的错误。

receiversList.paralelstream().map(t->{
                                      String receiverURL = serverURL + receiversList.get(i);
                                      HttpPost method = new HttpPost(receiverURL);

                                             String logPath = logFilesPath + logFilesList.get(i);
                                             return readMsg(logPath);
                                             })
                                     .flatMap(t->t.stream)
                                     .forEach(t->{
        StringEntity entity = new StringEntity(message);
                        log.info("Sending message:");
                        log.info(message + "\n");
                        method.setEntity(entity);
                        if (receiverURL.startsWith("https")) {
                            processAuthentication(method, username, password);
                        }
                        httpClient.execute(method).getEntity().getContent().close();})