消息的异步处理——使用哪些并发原语？

Question

我正在构建 PostgreSQL 有线协议的简单实现，并希望客户端向我的服务发送消息，然后在后台异步处理它们。我在理解何时使用 ExecutorService 与使用原始线程时遇到了一些麻烦。我正在使用两个BlockingQueues - 一个用于放置消息并将它们发送到服务器，另一个用于接收消息，到目前为止我的代码如下。

我想知道的是，在这里使用 ExecutorService 是否有意义，或者我应该创建并启动 ReceiveThread 和 SendThread 作为独立线程（即new Thread(new ReceiveThread()).start();）？

import java.io.*;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;


public class Connection {

    private MessageBuilder builder;
    private MessageReader reader;

    private BlockingQueue<byte[]> sendQueue;
    private BlockingQueue<byte[]> receiveQueue;

    private ExecutorService exec = Executors.newFixedThreadPool(2);

    private Socket socket;

    public Connection(String hostName, int port, String username) throws IOException {
        this(new MessageBuilder(), new MessageReader(), hostName, port, username);
    }

    public Connection(MessageBuilder builder, MessageReader reader, String hostName, int port, String username) throws IOException {
        this.builder = builder;
        this.reader = reader;
        this.sendQueue = new LinkedBlockingDeque<byte[]>();
        this.receiveQueue = new LinkedBlockingDeque<byte[]>();
        socket = new Socket(hostName, port);
        List<Param> params = new ArrayList<Param>();
        params.add(new Param("user", username));
        sendQueue.add(builder.buildStartupMessage(3, 0, params));

        exec.submit(new SendThread(sendQueue, new DataOutputStream(socket.getOutputStream())));
        exec.submit(new ReceiveThread(receiveQueue, new DataInputStream(socket.getInputStream()), new MessageReader()));
    }

    public void sendMessage(byte[] bytes) {
        sendQueue.add(bytes);
    }

    public void closeConnection() throws IOException {
        socket.close();
    }
}

class ReceiveThread implements Callable<Boolean> {
    private BlockingQueue<byte[]> receiveQueue;
    private DataInputStream dis;
    private MessageReader reader;

    public ReceiveThread(BlockingQueue<byte[]> queue, DataInputStream dis, MessageReader reader) {
        this.receiveQueue = queue;
        this.dis = dis;
        this.reader = reader;
    }

    public Boolean call() throws Exception {
        byte msgByte = dis.readByte();
        System.out.println("Response type is: " + (char) msgByte);
        int length = dis.readInt();
        byte[] message = new byte[length+1];
        message[0] = msgByte;
        byte[] bytes = ByteBuffer.allocate(4).putInt(length).array();
        System.arraycopy(bytes, 0, message, 1, bytes.length);
        int readLength = dis.read(message, 5, length - 5 );
        System.out.println("readLength : " + readLength + " should be length: " + (length-5));
        receiveQueue.put(message);
        return true;
    }
}

class SendThread implements Callable<Boolean> {

    private BlockingQueue<byte[]> sendQueue;
    private DataOutputStream dos;

    public SendThread(BlockingQueue<byte[]> queue, DataOutputStream dos) {
        this.sendQueue = queue;
        this.dos = dos;
    }

    public Boolean call() throws Exception {
        byte[] message = sendQueue.take();
        dos.write(message);
        return true;
    }
} 

Answer 1

使用ExecutorService 的主要好处是它提供了线程池。如果您有许多短期线程，则由创建这些线程引起的开销可能很大。 ExecutorService 可以通过使用线程池并将线程分配给您的任务来解决这个问题。它还负责调度。假设您有 15000 个工作要做，每个工作需要 5 毫秒才能完成。创建 15000 个线程显然没有意义。 ExecutorService 会将您的作业安排在 4 个线程上。您可以根据需要使用不同类型的ExecutorServices。

我不完全了解您的代码，为什么您不在发送和接收线程中使用循环。您的代码将仅接收/发送一条消息，我认为这不是您的意图。您可以摆脱 ExecutorService 并拥有 2 个长期存在的线程，一个用于发送，一个用于接收（就像您目前拥有的那样）。但是，在这些内部，您应该编写一个 while 循环来定期检查队列的内容。