Java 线程 - 等待数据返回，但不阻塞其他线程

Question

我有 7 个线程在 ExecutorPool 中运行，它们处理数据，并且偶尔需要来自另一个线程上运行的监听器实例的数据。监听器通过套接字向服务器发送请求，一段时间后，当返回结果时，监听器会将数据返回给调用它的工作线程。我想阻止工作线程直到返回请求的数据，但我不想阻止监听器从其他工作线程发出其他请求。我该怎么做？

Answer 1

如果一个线程将工作交给另一个线程，然后简单地等待结果，则不需要另一个线程来完成该工作。您可能需要一个完成这项工作的类，但它是在同一线程上调用的。如果多个线程使用同一个实例，则可能需要一些同步。但底线是这样的:

您不需要监听器线程。将其替换为处理请求的组件，并同步调用它。

编辑

根据您自己的回答，您的问题更加清楚了。正如@JimN建议的那样，您可能想要将一个Future分发给工作线程，并将其设为Listener保存在一个Map键控中的CompletableFuture通过请求 ID 直到响应返回。

示例代码:

public class WorkUnitProcessor implements Runnable {

    // ...

    @Override
    public void run() {
        while(true) {
            WorkUnit work = master.getNextWorkUnit();
            if(work == null) return;
            doWork(work);
        }
    }

    public void doWork(WorkUnit work) {

        //Do some work...

        try {
            DataRequest dataRequest = createRequest(work);

            Future<Response> future = server.getData(dataRequest);
            Response response = future.get();                       // this call blocks until the Response is available.

            //finish doing work

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            // handle e.getCause()
        }

    }

    // ...
}

public class Server implements DataSourceDrivenCallback {

    private final DataSource dataSource;

    private Map<Integer, CompletableFuture<Response>> openRequests = new ConcurrentHashMap<>();

    public Server(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public void incomingDataCallback(int requestId, ChunkOfData requestedData) {
        CompletableFuture<Response> responseHolder = openRequests.remove(requestId);  // get the responseHolder
        if (responseHolder != null) {
            responseHolder.complete(toResponse(requestedData));                     // make the response available.
        }
    }

    public Future<Response> getData(DataRequest datarequest) {
        int requestId = dataSource.submitRequest(serializeAndTranslateRequest(datarequest));
        CompletableFuture<Response> future = new CompletableFuture<>();
        openRequests.put(requestId, future);
        return future;
    }

    // ...
}