java - 内存消耗java文件流

Question

我正在通过 java nio 接口(interface)直接将文件读取到流中。这将启动异步 http 请求并在将来处理这些请求。每 10,000 条记录，我将此结果上传到服务器并清除记录，因此这清除了我的内存消耗。
我从字节数组开始，它一直保存在内存中。 http 客户端 (commons CloseableHttpAsyncClient) 触发异步请求，因此这些请求在开始时一次性触发。
有没有办法以一种可以限制同时处理的行数的方式来限制 lambda 流？从而控制我的内存。

new BufferedReader(new InputStreamReader(new ByteArrayInputStream(file)))
    .lines()
    .map(line -> CsvLine.create(line))
    .filter(line -> !line.isHeader())
    .forEach(line -> getResult(line, new FutureCallback<HttpResponse>() {
        @Override
        public void completed(HttpResponse response) {
            try {
                result.addLine(response);
            } catch (IOException e) {
                LOGGER.error("IOException, cannot write to server", e);
                todo.set(-1); // finish in error
            } finally {
                todo.decrementAndGet();
            }
       }

       @Override
       public void failed(Exception ex) {
           handleError();
       }

       @Override
       public void cancelled() {
           handleError();
       }
    }
));

Answer 1

您可以尝试使用信号量来限制您的流，以便一次只有特定的最大异步请求未完成。它可能看起来像这样:

Semaphore semaphore = new Semaphore(MAX_CONCURRENT_REQUESTS, true); // false if FIFO is not important
new BufferedReader(new InputStreamReader(new ByteArrayInputStream(file)))
.lines()
        .map(line -> CsvLine.create(line))
        .filter(line -> !line.isHeader())
        .forEach(line -> {
            try {
                if (!semaphore.tryAcquire(ASYNC_REQUEST_TIMEOUT, TimeUnit.MILLISECONDS)) {
                    handleTimeout();
                } else {
                    getResult(line, new FutureCallback<HttpResponse>() {
                        @Override
                        public void completed(HttpResponse response) {
                            try {
                                result.addLine(response);
                            } catch (IOException e) {
                                LOGGER.error("IOException, cannot write to server", e);
                                todo.set(-1); // finish in error
                            } finally {
                                todo.decrementAndGet();
                                semaphore.release();
                            }
                        }

                        @Override
                        public void failed(Exception ex) {
                            handleError();
                            semaphore.release();
                        }

                        @Override
                        public void cancelled() {
                            handleError();
                            semaphore.release();
                        }
                    }
                    );
                }
            } catch (InterruptedException e) {
                // handle appropriately
            }

        });