从 0 到 1 ,手把手教你编写《消息队列》项目(Java实现) —

文章目录

一、自定义应用层协议
- 请求与响应
- channel
二、自定义请求格式
三、自定义响应格式
四、服务器代码编写

一、自定义应用层协议

咱们这里的客户端与服务器的通信是基于TCP协议实现的.

当前要交互的 Message,以及调用各种API的请求,其实都是二进制数据.

因此咱们要自定义一个应用层协议(格式)来规范这些数据.

请求与响应

咱们规定以下格式来表示请求与响应.
在这里插入图片描述
Type 表示当前这个请求和响应是做什么的.
Length 表示接下来的Payload的长度
Payload 才是真正需要用到的数据.

客户端与服务器之间要进行的操作,也就是咱们虚拟机提供的核心API,
而此处的Type就与这些API一一对应,故而也就可以通过Type来告诉服务器客户端调用的是哪个API.

0x1 创建channel
0x2 关闭channel
0x3 创建 exchange
0x4 销毁 exchange
0x5 创建 queue
0x6 销毁 queue
0x7 创建 binding
0x8 销毁 binding
0x9 发送 message
0xa 订阅 message
0xb 返回 ack(手动应答消息)
0xc 服务器给客户端推送消息(自动推送消息的独有响应)

channel

channel 是什么呢?

channel其实只是咱们逻辑上的一个信道,TCP的连接是需要三次握手的,
因此咱们为了尽量减少短时间多次发送请求时,用于建立TCP连接的资源消耗,
故而咱们就建立一个逻辑上的信道,让这些信道去共享一个TCP连接,
达到对这一个TCP连接的复用,从而减少多次建立TCP连接的资源消耗.

二、自定义请求格式

如果是请求,以创建 exchange 举例:
Type:0x3
Length:Payload的长度
Payload:要创建的交换机的参数列表

客户端可能会同时发送多个请求,创建 exchange,创建 queue,创建 binding等.
那么我们就要想办法将响应与请求对应起来,

所以此处咱们规定,所有请求的 Payload中都必须包含的两个字段:

String channelId (这次通信使用的 channel 的身份标识)
String RId (表示一次请求/响应的身份标识，可以把请求和响应对上)

这里不同的请求Payload这个参数列表又有所不同,

故而咱们需要创建一个包来专门写各个请求对应的类
在这里插入图片描述

/**
 * 表示一个网络通信中的请求对象，按照自定义协议的格式来展开的
 */
@Data
public class Request {
    private int type;
    private int length;
    private byte[] payload;
}

/**
 * 使用这个类表示请求方法Payload的公共参数/辅助的字段
 * 后续每个方法又会有一些不同的参数，不同的参数再分别使用不同的子类来表示
 */
@Data
public class BasicArguments implements Serializable {
    // 表示一次请求/响应 的身份标识，可以把请求和响应对上
    protected String rid;
    // 这次通信使用的 channel 的身份标识
    protected String channelId;
}

/**
 * 创建交换机的请求的Payload字段类
 */
@Data
public class ExchangeDeclareArguments extends BasicArguments implements Serializable {
    private String exchangeName;
    private ExchangeType exchangeType;
    private boolean durable;
    private boolean autoDelete;
    Map<String,Object> arguments;
}

/**
 * 销毁交换机的请求的Payload字段类
 */
@Data
public class ExchangeDeleteArguments extends BasicArguments implements Serializable {
    private String exchangeName;
}

/**
 * 创建队列的请求的Payload字段类
 */
@Data
public class QueueDeclareArguments extends BasicArguments implements Serializable {
    private String queueName;
    private boolean durable;
    private boolean exclusive;
    private boolean autoDelete;
    private Map<String,Object> arguments;
}

/**
 * 销毁队列的请求的Payload字段类
 */
@Data
public class QueueDeleteArguments extends BasicArguments implements Serializable {
    private String queueName;
}

/**
 * 创建 绑定关系的请求的Payload字段类
 */
@Data
public class BindingDeclareArguments extends BasicArguments implements Serializable {
    private String exchangeName;
    private String queueName;
    private String bindingKey;
}

/**
 * 解除绑定关系的请求的Payload字段类
 */
@Data
public class BindingDeleteArguments extends BasicArguments implements Serializable {
    private String exchangeName;
    private String queueName;
}

/**
 * 发布消息的请求的Payload字段类
 */
@Data
public class BasicPublishArguments extends BasicArguments implements Serializable {
    private String exchangeName;
    private String routingKey;
    private BasicProperties basicProperties;
    private byte[] body;
}

/**
 * 订阅消息的请求的Payload字段类
 */
@Data
public class BasicConsumeArguments extends BasicArguments implements Serializable {
    private String consumerTag;
    private String queueName;
    private boolean autoAck;
    // 这个类对应的 basicConsume 方法中，还有一个参数，是回调函数(如何来处理消息)
    // 这个回调函数，是不能通过网络传输的
    // 站在 broker server 这边，针对消息的处理回调，其实是统一的(把消息返回给客户端)
    // 客户端收到消息之后，再在客户端自己这边执行一个用户自定义的回调就行了
    // 此时，客户端也就不需要把自身的回调告诉给服务器了
    // 所以不需要 consumer 这个成员
}

/**
 * 手动消息应答的请求的Payload字段类
 */
@Data
public class BasicAckArguments extends BasicArguments implements Serializable {
    private String queueName;
    private String messageId;
}

三、自定义响应格式

如果是响应,以创建 exchange 成功的响应举例:
Type:0x3
Length:Payload的长度
Payload:我们可以自行规定.

客户端可能会同时发送多个请求,创建 exchange,创建 queue,创建 binding等.
那么我们就要想办法将响应与请求对应起来,

所以此处咱们规定,所有请求的 Payload中都必须包含的两个字段:

String channelId (这次通信使用的 channel 的身份标识)
String RId (表示一次请求/响应的身份标识，可以把请求和响应对上)

此处咱们就要想,咱们响应其实大致就分为两种,
一种是客户端调用服务器的API(创建 exchange等),此时我们只需要在响应中添加 boolean ok 这个字段来告诉客户端方法是否执行成功就可以.

另一种是服务器主动向客户端发送的消息,此时就必须要含有消息的具体数据了.

所以也要创建一个包来专门写响应类
在这里插入图片描述

/**
 * 这个对象表示一个响应，也是根据自定义应用层协议来的
 */
@Data
public class Response {
    private int type;
    private int length;
    private byte[] payload;
}

/**
 * 这个类表示各个远程调用的方法的返回值的公共信息
 */
@Data
public class BasicReturns implements Serializable {
    // 表示一次请求/响应 的身份标识，可以把请求和响应对上
    protected String rid;
    // 这次通信使用的 channel 的身份标识
    protected String channelId;
    // 表示当前这个远程调用方法的返回值
    protected boolean ok;
}

/**
 * 将消息自动发送给消费者的响应类
 */
@Data
public class SubScribeReturns extends BasicReturns implements Serializable {
    private String consumerTag;
    private BasicProperties basicProperties;
    private byte[] body;
}

四、服务器代码编写

/**
 * 这个 BrokerServer 就是咱们 消息队列 本体服务器
 * 本质上就是一个 TCP 服务器
 */
public class BrokerServer {
    private ServerSocket serverSocket = null;

    // 当前考虑一个 BrokerServer 上只有一个 虚拟主机
    private VirtualHost virtualHost = new VirtualHost("default");

    // 使用这个 哈希表 表示当前的所有会话(也就是说有哪些客户端正在和咱们的服务器进行通信)
    // 此处的 key 是 channelId，value 为对应的 Socket 对象
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();

    // 引入一个线程池，来处理多个客户端的请求
    private ExecutorService executorService = null;

    // 引入一个 boolead 变量控制服务器是否继续运行
    private volatile boolean runnable = true;

    // 构造方法,指定程序运行的端口号
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    // 启动服务器的方法
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动!");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                Socket clientSocket = serverSocket.accept();
                // 把处理连接的逻辑丢给这个线程池
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        }catch (SocketException e) {
            System.out.println("[BrokerServer] 服务器停止运行");
        }

    }

    // 一般来说停止服务器，就是直接 kill 掉对应进行就行了
    // 此处还是搞一个单独的停止方法，主要是用于后续的单元测试
    public void stop() throws IOException {
        runnable = false;
        // 把线程池中的任务都放弃了，让线程都销毁
        executorService.shutdownNow();
        serverSocket.close();
    }

    // 通过这个方法，来处理一个客户端的连接
    // 在这一个连接中，可能会涉及到多个请求和响应
    private void processConnection(Socket clientSocket) {
        try(InputStream inputStream = clientSocket.getInputStream();
            OutputStream outputStream = clientSocket.getOutputStream()) {
            // 这里需要按照特定格式来读取并解析，此时就需要用到 DataInputStream 与 DataOutputStream
            try(DataInputStream dataInputStream = new DataInputStream(inputStream);
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
                    // 1.读取请求并解析
                    Request request = readRequest(dataInputStream);
                    // 2.根据请求计算响应
                    Response response = process(request,clientSocket);
                    // 3.把响应写回给客户端
                    writeResponse(response,dataOutputStream);
                }
            }
        } catch (EOFException | SocketException e) {
            // 对于这个代码，DataInputStream 如果读到 EOF，就会抛出一个 EOFException 异常
            // 需要借助这个异常来结束循环
            System.out.println("[BrokerServer] connection 关闭! 客户端的地址: " + clientSocket.getInetAddress().toString() + "客户端的端口号" +
                    clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e) {
            e.printStackTrace();
        } finally {
            try {
                // 当连接处理完了，记得关闭 socket
                clientSocket.close();
                // 一个 TCP 连接中，可能包含多个 channel， 需要把当前这个 socket 对应的所以 channel 也顺便清理掉
                clearClosedSession(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    // 用来读取客户端的请求
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payload = new byte[request.getLength()];
        int n = dataInputStream.read(payload);
        if (n != request.getLength()) {
            throw new IOException("读取请求格式错误");
        }
        request.setPayload(payload);
        return  request;
    }

    // 用来向客户端写入响应
    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 这个刷新缓冲区也是重要操作!!!
        dataOutputStream.flush();
    }

    // 处理请求并构造响应
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 1.把 request 中的 payload 做一个初步解析,得到 channelId与Rid
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid=" + basicArguments.getRid() + ", channelId=" + basicArguments.getChannelId()
                + ", type=" + request.getType() + ", length=" + request.getLength());
        // 2.根据 type 的值，来调用对应的API
        boolean ok = true;
        if (request.getType() == 0x1) {
            // 创建 channel
            sessions.put(basicArguments.getChannelId(), clientSocket);
            System.out.println("[BrokerServer] 创建 channel 完成! channelId=" + basicArguments.getChannelId());
        } else if (request.getType() == 0x2){
            // 销毁 channel
            sessions.remove(basicArguments.getChannelId());
            System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" + basicArguments.getChannelId());
        } else if (request.getType() == 0x3){
            // 创建 Exchange,此时 payload 就是 ExchangeDeclarArguments 对象了
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                    arguments.isDurable(),arguments.isAutoDelete(),arguments.getArguments());
            System.out.println("[BrokerServer] 创建 exchange 完成! exchangeName=" + (arguments.getExchangeName()));
        } else if (request.getType() == 0x4){
            // 销毁交换机
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
            System.out.println("[BrokerServer] 销毁 exchange 完成! exchangeName=" + (arguments.getExchangeName()));
        } else if (request.getType() == 0x5){
            // 创建队列
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueuDeclare(arguments.getQueueName(), arguments.isDurable(),
                    arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());
            System.out.println("[BrokerServer] 创建 queue 完成! queueName=" + (arguments.getQueueName()));
        } else if (request.getType() == 0x6){
            // 销毁队列
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete(arguments.getQueueName());
            System.out.println("[BrokerServer] 销毁 queue 完成! queueName=" + (arguments.getQueueName()));
        } else if (request.getType() == 0x7){
            // 创建绑定
            BindingDeclareArguments arguments = (BindingDeclareArguments) basicArguments;
            ok = virtualHost.bindingDeclare(arguments.getExchangeName(), arguments.getQueueName(),
                    arguments.getBindingKey());
            System.out.println("[BrokerServer] 创建 绑定 完成! " + "exchangeName=" + (arguments.getExchangeName()) + "queueName=" + (arguments.getQueueName()));
        } else if (request.getType() == 0x8){
            // 解除绑定
            BindingDeleteArguments arguments = (BindingDeleteArguments) basicArguments;
            ok = virtualHost.bindingDelete(arguments.getExchangeName(), arguments.getQueueName());
            System.out.println("[BrokerServer] 销毁 绑定 完成! " + "exchangeName=" + (arguments.getExchangeName()) + "queueName=" + (arguments.getQueueName()));
        } else if (request.getType() == 0x9){
            // 发布消息
            BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(),
                    arguments.getBasicProperties(), arguments.getBody());
            System.out.println("[BrokerServer] 发布消息 成功! exchangeName=" + (arguments.getExchangeName()));
        } else if (request.getType() == 0xa){
            // 订阅消息
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(), new Consumer() {
                // 这里直接将回调函数写死,
                // 这个回调函数要做的工作，就是把服务器收到的消息可以直接推送回对应的消费者客户端
                @Override
                public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                    // 先知道当前这个收到的消息，要发送给哪个客户端
                    // 此处 consumerTag 其实是 channelId 根据 channelId 去 sessions 中查询，就可以得到对应的socket 对象，
                    // 从而发送数据
                    // 1.根据 channelId 找到 socket 对象
                    Socket clientSocket = sessions.get(consumerTag);
                    if (clientSocket == null || clientSocket.isClosed()) {
                        throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭");
                    }


                    // 2.构造响应数据
                    SubScribeReturns subScribeReturns = new SubScribeReturns();
                    subScribeReturns.setConsumerTag(consumerTag);
                    subScribeReturns.setChannelId(consumerTag);
                    subScribeReturns.setRid(""); // 这个回调函数是向客户端发送消息时触发的，并没有请求需要去对应，因此可以不设置
                    subScribeReturns.setOk(true);
                    subScribeReturns.setBasicProperties(basicProperties);
                    subScribeReturns.setBody(body);

                    // 将响应对象 序列化
                    byte[] payload = BinaryTool.toBytes(subScribeReturns);
                    Response response = new Response();
                    // 0xc 表示服务器给消费者客户端推送的消息数据
                    response.setType(0xc);
                    response.setLength(payload.length);
                    // response 的 payload 就是一个 SubScribeReturns
                    response.setPayload(payload);


                    // 3.发送响应到客户端
                    // 注意! 此处的 dataOutputStream 这个对象不能 close !!!
                    // 如果把 dataOutputStream 关闭，就会直接把 clientSocket 里的 outputStream 也关了
                    // 此时就无法继续向该客户端发送其他消息了
                    writeResponse(response,new DataOutputStream(clientSocket.getOutputStream()));
                }
            });
            System.out.println("[BrokerServer] 订阅消息 成功! queueName=" + (arguments.getQueueName()));
        } else if (request.getType() == 0xb){
            // 返回 ack
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
            System.out.println("[BrokerServer] 消息应答 成功! MessageId=" + (arguments.getMessageId()));

        } else {
            // 当前的 type 是非法的
            throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
        }
        // 构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toBytes(basicReturns);

        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Response] rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId());
        return response;
    }

    // 销毁所有的 channel信道
    private void clearClosedSession(Socket clientSocket) {
        // 这里要做的事情，主要就是遍历上述 session hash 表，把该关闭的 socket 对应的键值对，统统删掉
        List<String> toDeleteChannelId = new ArrayList<>();
        for (Map.Entry<String,Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId : toDeleteChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);
    }

}