java - 多个线程读取和导入目录中的文件

Question

我有一个 Java 程序要执行 2 个步骤:

1. 使用多个线程递归读取目录中的文件(10线程)
2. 通过 HTTP Post 将这些文件发送到服务器

似乎第一步运行良好，但第二步总是发送同一个文件 10 次。

如何纠正这个错误？

这是我的日志:

Import file: 2005_1.xml
Import file: 2005_7.xml
Import file: 2005_6.xml
Import file: 2005_10.xml
Import file: 2005_5.xml
Import file: 2005_11.xml
Import file: 2005_8.xml
Import file: 2005_2.xml
Import file: 2005_3.xml
Import file: 2005_4.xml

Result: {"fileName":"2005_4.xml"
Result: {"fileName":"2005_4.xml"
...

Response: HttpResponseProxy{HTTP/1.1 400  [
Import file: 2005_9.xml
Result: {"fileName":"2005_4.xml"
...
Response: HttpResponseProxy{HTTP/1.1 200  [
Result: {"fileName":"2005_4.xml"

Response: HttpResponseProxy{HTTP/1.1 200  [
Result: {"fileName":"2005_9.xml"

还有我的代码:多线程读取目录下的文件:

public void listSendFilesMultiThread(final File folder) {
        ExecutorService service = Executors.newFixedThreadPool(10, getThreadFactory());

        for (final File fileEntry : folder.listFiles()) {
            Runnable r;
            r = new Runnable() {
                @Override
                public void run() {
                    if (fileEntry.isDirectory()) {
                        listSendFilesMultiThread(fileEntry);
                    } else {
                        GetThread thread = new GetThread(fileEntry, errorFilesDestDir);

                        // start the thread
                        thread.start();

                        // join the threads
                        try {
                            thread.join();
                        } catch (InterruptedException e) {
                            LOGGER.error("InterruptedException: " + e);
                        }
                    }
                }
            };
            service.execute(r);
        }
    }

通过 HTTP Post 发送文件:

static class GetThread extends Thread {

        private final CloseableHttpClient closeableHttpClient;
        private final File file;
        private final String errorFilesDestDir;
        private final PoolingHttpClientConnectionManager cm;
        private MultipartEntityBuilder builder;

        public GetThread(File file, String errorFilesDestDir) {
            cm = new PoolingHttpClientConnectionManager();
            closeableHttpClient = HttpClients.custom().setConnectionManager(cm).build();

            this.file = file;
            this.errorFilesDestDir = errorFilesDestDir;
        }

        @Override
        public void run() {
            try {
                if (file.exists() && file.length() > 0) {
                    FileBody fileBody = new FileBody(file);
                    // we should create a new builder per file
                    builder = MultipartEntityBuilder.create();
                    builder.setMode(HttpMultipartMode.BROWSER_COMPATIBLE);
                    builder.addPart("xmlFile", fileBody);
                    HttpEntity entity = builder.build();
                    request.setEntity(entity);

                    LOGGER.info("Import file: " + file.getName());
                    CloseableHttpResponse response = closeableHttpClient.execute(request);
                    LOGGER.info("Response: {}", response.toString());

                    try {
                        entity = response.getEntity();
                        printInfo(response, entity);
                    } finally {
                        response.close();
                        closeableHttpClient.close();
                        cm.close();
                    }

                    EntityUtils.consume(entity);
                } else if (file.length() == 0) {
                    LOGGER.error("The import XML file is empty: " + file.getAbsolutePath());
                    Files.copy(file.toPath(), new File(errorFilesDestDir + file.getName()).toPath(), StandardCopyOption.REPLACE_EXISTING);
                } else {
                    LOGGER.error("The import XML file doesn't exist");
                }
            } catch (ClientProtocolException e) {
                // Handle protocol errors
                LOGGER.error("ClientProtocolException: " + e.toString());
            } catch (IOException e) {
                // Handle I/O errors
                LOGGER.error("IOException: " + e.toString());
            }
        }
    }

连接驱逐策略

public static class IdleConnectionMonitorThread extends Thread {

        private final HttpClientConnectionManager connMgr;
        private volatile boolean shutdown;

        public IdleConnectionMonitorThread(HttpClientConnectionManager connMgr) {
            super();
            this.connMgr = connMgr;
        }

        @Override
        public void run() {
            try {
                while (!shutdown) {
                    synchronized (this) {
                        wait(5000);
                        // Close expired connections
                        connMgr.closeExpiredConnections();
                        // Optionally, close connections
                        // that have been idle longer than 30 sec
                        connMgr.closeIdleConnections(30, TimeUnit.SECONDS);
                    }
                }
            } catch (InterruptedException e) {
                LOGGER.error("InterruptedException: " + e.toString());
            }
        }

        public void shutdown() {
            shutdown = true;
            synchronized (this) {
                notifyAll();
            }
        }
    }

Answer 1

我建议使用生产者和消费者模式来简化代码。一个或多个线程(一个应该足够了，因为它不做任何处理)将使用您的逻辑并找到要上传的文件。接下来，它会将它们放入队列中。启动任意数量的消费者，从队列中读取记录并将它们上传到服务器。 https://dzone.com/articles/concurrency-pattern-producer消费者将具有从磁盘读取文件并上传到服务器的逻辑。