java - 我可以使用哪种 Java 线程池来处理来自套接字的数据？

Question

我有以下用于启动线程服务器的代码:

Thread server = new Thread(new ServerRunnable(serverPort, devMode, messageQueue, database));
server.start();

Thread worker1 = new Thread(
    new WorkerRunnable(
        messageQueue,
        database, 
        devMode, 
        1
    ));
Thread worker2 = new Thread(
    new WorkerRunnable(
        messageQueue,
        database, 
        devMode, 
        2
    ));
Thread worker3 = new Thread(
    new WorkerRunnable(
        messageQueue,
        database, 
        devMode, 
        3
    ));

worker1.start();
worker2.start();
worker3.start();

ServerRunnable将包含读取字节和其他信息的 State 对象传递到 messageQueue 。WorkerRunnable线程获取消息并处理它们。

我正在查看ThreadPoolExecutor ，希望我可以用它来用一个可以根据需要增长或缩小的池来替换上面的三个工作线程，但它并没有按照我预期的方式工作。它需要一个任务队列来完成。

我的代码使用 Java NIO，因此无法保证放入队列的项目是完整的，因此可能需要进一步处理。结果我无法使用 ThreadPoolExecutor按照我最初想象的方式，这将通过 Runnable包含队列。

所以，我突然想到如果我想使用ThreadPoolExecutor这里我必须将其添加到ServerRunnable中类(有点反转我当前的流程)，并传递一个新的 WorkerRunnable类上 ThreadPoolExecutor将消息参数传递给队列后。那是对的吗？

可能是这样的:

LinkedBlockingQueue messageQueue = new LinkedBlockingQueue<Runnable>();

Thread server = new Thread(
    new ServerRunnable(
        serverPort, 
        devMode, 
        messageQueue, 
        database
    ));
server.start();
int  corePoolSize  =    5;
int  maxPoolSize   =   10;
long keepAliveTime = 5000;

ExecutorService threadPoolExecutor = 
    new ThreadPoolExecutor(
        corePoolSize,
        maxPoolSize,
        keepAliveTime,
        TimeUnit.MILLISECONDS,
        messageQueue); 
threadPoolExecutor.prestartAllCoreThreads()

然后在 ServerRunnable类:

// Process incoming bytes into a message

messageQueue.put(new WorkerRunnable(database, devMode, message));

我的理解正确吗？

Answer 1

是的，或者您可以在队列中拥有单个消费者，该消费者将继续将任务传递给 ExecutorService。像这样的事情

Thread consumer = new Thread(new Consumer(queue));
consumer.start();

class Consumer implements Runnable {
    private ExecutorService     service = Executors.newCachedThreadPool();
    private final BlockingQueue queue;

    public Consumer(Queue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        Task t = null;
        while(t = queue.take()) {
            Worker worker = new Worker(t);
            service.execute(worker);
        }
    }
}

这与 ExecutorService 的耦合有点松散。