消息队列（六）：服务器设计

紧接着上一章没说完的进行服务器的补充。

推送给消费者消息的基本实现思路

1. 让 brokerServer 把哪些消费者管理好
2. 收到对应的消息，把消息推送给消费者

消费者是以队列为维度来订阅消息的，一个队列可以有多个消费者（此处我们约定按照轮询的方式来进行消费）

消费者消费消息的核心逻辑

这里又一次提到了消费者，我们来把消费者相关的代码完善一下。

消费者管理实现

消费者

先前我们提到了这个函数式接口，这个接口的作用就是用来消费消息（和消费者作用一样）。

那么我们就给这个函数式接口起名消费者 consumer 

上一章也提到了具体的代码，这里再演示一次：

@FunctionalInterface
public interface Consumer {
    // Deliver 的意思就是"投递"，这个方法预期是在每次服务器收到消息，来调用
    // 通过这个方法把消息推送给对应的消费者
    void handleDeliver(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException;
 
}

具体是想干啥，谁调用由谁决定（具体代码）。

消费者完整的执行环境

消费者在消费的时候，需要知道是哪个消费者进行消费了，是哪个队列传过来的消息，知否自动应答。

故此，我们大概有如下几个参数：

   private String consumerTag;
   private String queueName;
   private boolean autoAck;
   // 通过这个回调来处理收到的消息
   private Consumer consumer;

以及对应的 getting、setting  方法 和 构造方法 。

消费者管理类

这个类我放在了核心类。

啥时候执行这个消费者完整的执行环境啊，通过这个类，来实现消费消息的核心逻辑。

订阅消息的核心 就是这个    consumer.addConsumer() 。

根据这个图，我们也能看出来我们大致需要如下几个属性：

需要一个队列，一个扫描线程，此外还需要记录一下是哪个虚拟主机持有的消费者；此外，还需要一个执回调（函数式接口）的线程池。

    // 持有上层的 VirtualHost 对象的引用. 用来操作数据.
    private VirtualHost parent;
    // 指定一个线程池, 负责去执行具体的回调任务.
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scannerThread = null;

方法：

• 构造方法
• 添加消费者
• 消费消息
• 唤醒消费

先来说说构造方法：

构造器只持有虚拟机的名字；因为扫描线程是不断地进行扫描，啥时候启动这个扫描线程呢？这里就设置在了构造方法中。

扫描线程的实现逻辑就如上图所言，一旦调用虚拟主机中的发送消息就会唤醒消费（具体的唤醒就是往存放令牌的队列中添加队列名），扫描线程扫描的就是这个队列，一旦队列有消息进来就调用消费消息这个方法。

public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
 
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 根据令牌找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在！queueName="+queueName);
                    }
                    // 3. 从这个队列中消费一个信息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                }catch (InterruptedException | MqException e) {
                    e.printStackTrace();;
                }
            }
        });
        // 将线程设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

唤醒消费

该方法只有一行代码，就是将队列名放入令牌队列。

    // 这个方法的调用时机就是发送消息的时候.
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

添加消费者

在这个方法中，之前记录的虚拟机主机名就派上用场了。

我们需要从虚拟主机中获取到队列（调用者需要传递下来队列名称），类型为 核心类： MSGQueue（此时可以补充两个方法）：

// 添加一个新的订阅者
    public void addConsumerEnv(ConsumerEnv consumerEnv) {
        consumerEnvList.add(consumerEnv);
    }
    // 订阅者删除暂时先不考虑
    // 先挑选一个订阅者，用来处理当前的消息（轮询的方式）
    public ConsumerEnv chooseConsumer() {
        if (consumerEnvList.size() == 0) {
            // 该队列没有人订阅
            return null;
        }
        // 计算一下当前要取的元素下标
        int index = consumerSeq.get() % consumerEnvList.size();
        consumerSeq.getAndIncrement();
        return consumerEnvList.get(index);
    }

我们采取一个轮询的方式去处理消息，每个消费者都有机会进行消费消息。

具体的消费就是通过下标取模的方式，这里涉及到了多线程同时调用，所以使用了原子类 AtomicInteger 修饰 consumerSeq 。

没有就需要抛出异常，有的话需要创建出完整的消费者环境（将参数都传进去），随后将其添加到队列中去。

在进行循环，如果队列有消息就先进行消费完。

具体代码如下：

public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前这个队列已经有消息了，就需要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 这个放啊调用一次就消费一条消息
                consumeMessage(queue);
            }
        }
    }

消费消息

大致逻辑

1. 按照轮询的方法，找个消费者出来（没有消费者就暂时不消费，等有消费者出现才进行消费）
2. 从队列中取出一个消息（当前队列中还没有消息，也不需要消费）
3. 把消息带入到消费者的回调方法中, 丢给线程池执行
   1. 把消息放到待确认的集合中. 这个操作势必在执行回调之前
   2. 真正执行回调操作
   3. 如果当前是 "自动应答" , 就可以直接把消息删除了. 如果当前是 "手动应答" , 则先不处理, 交给后续消费者调用 basicAck 方法来处理.
      1. 删除硬盘上的消息
      2. 删除上面的待确认集合中的消息
      3. 删除内存中消息中心里的消息

具体代码如下：

private void consumeMessage(MSGQueue queue) {
        // 1. 按照轮询的方法，找个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前队列没有消费者, 暂时不消费. 等后面有消费者出现再说.
            return;
        }
        // 2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 当前队列中还没有消息, 也不需要消费.
            return;
        }
        // 3. 把消息带入到消费者的回调方法中, 丢给线程池执行.
        workerPool.submit(() -> {
            try {
                // 1. 把消息放到待确认的集合中. 这个操作势必在执行回调之前.
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 真正执行回调操作
                luckyDog.getConsumer().handleDeliver(luckyDog.getConsumerTag(), message.getBasicProperties(),
                        message.getBody());
                // 3. 如果当前是 "自动应答" , 就可以直接把消息删除了.
                //    如果当前是 "手动应答" , 则先不处理, 交给后续消费者调用 basicAck 方法来处理.
                if (luckyDog.isAutoAck()) {
                    // 1) 删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中消息中心里的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
 
    }
}

关于确认消息

能够确保消息是被正确的消费掉了，消费者的回调函数，顺利执⾏完了（中间没有抛出异常）

这条消息就可以被删除了。

消息确认也就是为了保证“消息不丢失”

为了达成消息不丢失这样的效果，这样处理：

• 在真正执⾏回调之前，把这个消息先放到 “待确认的集合”中
• 真正回调
• 当前消费者采取的是 autoAck=true,就认为回调执⾏完毕不抛异常，就算消费成功，然后就可以删除消息
• 当前消费者采取的是 autoAck=false，⼿动应答，就需要消费者再回调函数内部，显式调⽤
  basicAck这个核⼼API

basicAck实现原理，⽐较简单，当传⼊参数 autoAck=false, 就⼿动再回调函数的时候，调⽤

basicAck 就⾏（具体的在 VirtualHost中）

消息确认是为了保证消息不丢失，而需要的逻辑。

• 1. 执⾏回调⽅法的过程中，抛异常了
  • 当回调函数异常，后续逻辑执⾏不到了。此时这个消费就会始终待在待确认集合中。
    RabbitMQ中会设置⼀个死信队列，每⼀个队列都会绑定⼀个死信队列。应⽤场景：当消息在 消费过程中出现异常，就会把消息投⼊到死信队列中；当消息设置了过期时间，如果在过期时 间内，没有被消费，就会投⼊到死信队列中；当队列达到最⼤⻓度时，新的消息将⽆法被发送 到队列中。此时，RabbitMQ可以选择将这些⽆法发送的消息发送到死信队列中，以便进⾏进 ⼀步处理
• 2. 执⾏回调过程中, Broker Server崩溃了，内存数据都没了！但是硬盘数据还在，正在消费的这个 消息，在硬盘中仍然存在。BrokerServer重启后，这个消息就⼜被加载到内存了，就像从来没被消 费过⼀样。消费者就会有机会重新得到这个消息。

BrokerServer

Broker Server 本质是一个服务器，我在这个自定义服务器上添加了自定义应用层协议。

自定义协议

具体的协议设置：

请求和响应

•  type ： 用于描述当前这个请求和响应是要干啥的
  • 在MQ中，客户端（⽣产者 + 消费者）和 服务器 （Broker Server）之间，要进⾏哪些操作？（就是VirtualHost中的那些核⼼API）
  • 希望客户端，能通过⽹络远程调⽤这些API
  • 此处的type就是描述当前这个请求/响应是在调⽤哪个API
  • 
  • TCP是有连接的，Channel 是 Connection 内部的逻辑连接。此时⼀个 Connection 中可能有多 个连接，为啥要这么设计？就是为了让 TCP 连接得到复用（不断地创建和删除 TCP 连接，成本还是比较高的）
• length：⾥⾯存储的是 payload的⻓度。⽤4个字节来存储
• payload：会根据当前是请求还是响应，以及当前的 type 有不同的值
  • ⽐如 type 是 0x3(创建交换机)，同时当前是个请求，此时 payload 的内容，就相当于是

    exchangeDelcare 的参数的序列化结果
  • ⽐如 type 是 0x3(创建交换机)，同时当前是个响应，此时 payload 的内容，就相当于是
    exchangeDelcare 的返回结果的序列化内容

每一个请求对应的 响应不同（重点是 payload 不同），所以对应每一个请求都单独设计一个类，帮助构造响应。

ExchangeDelcare

request

response

通信流程：

由于不同的 payload 我们需要对其进行设计：

根据上述图示，我们需要如下几个参数：

由于每次响应都会带有 rid 和 channelId，所以将其设为父类：

其他的类也一样，继承这个类，并实现串行化，我就不一一举例了，我把大致的图放下来：

ExchangeDelete

QueueDelcare

QueueDelete

QueueBind

QueueUnBind

BasicPublish

BasicConsumer

BasicAck

创建 BrokerServer类

消息队列本体服务器

实现读取请求和写回响应

读取请求

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(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 这个刷新缓冲区也是重要的操作!!
        dataOutputStream.flush();
    }

清理过期会话

private void clearClosedSession(Socket clientSocket) {
        // 这里要做的事情, 主要就是遍历上述 sessions hash 表, 把该被关闭的 socket 对应的键值对, 统统删掉.
        List<String> toDeleteChannelId = new ArrayList<>();
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                // 不能在这里直接删除!!!
                // 这属于使用集合类的一个大忌!!! 一边遍历, 一边删除!!!
                // sessions.remove(entry.getKey());
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId : toDeleteChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);
    }

解析请求（Process）

这一步属于重中之重。

大致流程如下：

1. 把 request 中的 payload 做一个初步解析
2. 根据 type 的值，决定具体要干啥
3. 构造响应

private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 1. 把 request 中的 payload 做一个初步的解析.
        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 的值, 来进一步区分接下来这次请求要干啥.
        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) {
            // 创建交换机. 此时 payload 就是 ExchangeDeclareArguments 对象了.
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                    arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
        } else if (request.getType() == 0x4) {
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
        } else if (request.getType() == 0x5) {
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(),
                    arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());
        } else if (request.getType() == 0x6) {
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete((arguments.getQueueName()));
        } else if (request.getType() == 0x7) {
            QueueBindArguments arguments = (QueueBindArguments) basicArguments;
            ok = virtualHost.queueBind(arguments.getQueueName(), arguments.getExchangeName(), arguments.getBindingKey());
        } else if (request.getType() == 0x8) {
            QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;
            ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());
        } else if (request.getType() == 0x9) {
            BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(),
                    arguments.getBasicProperties(), arguments.getBody());
        } else if (request.getType() == 0xa) {
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                    new Consumer() {
                        @Override
                        public void handleDeliver(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.setChannelId(consumerTag);
                            subScribeReturns.setRid(""); // 由于这里只有响应, 没有请求, 不需要去对应. rid 暂时不需要.
                            subScribeReturns.setOk(true);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            byte[] payload = BinaryTool.toBytes(subScribeReturns);
                            Response response = new Response();
                            // 0xc 表示服务器给消费者客户端推送的消息数据.
                            response.setType(0xc);
                            // response 的 payload 就是一个 SubScribeReturns
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            // 3. 把数据写回给客户端.
                            //    注意! 此处的 dataOutputStream 这个对象不能 close !!!
                            //    如果 把 dataOutputStream 关闭, 就会直接把 clientSocket 里的 outputStream 也关了.
                            //    此时就无法继续往 socket 中写入后续数据了.
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(dataOutputStream, response);
                        }
                    });
        } else if (request.getType() == 0xb) {
            // 调用 basicAck 确认消息.
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
        } else {
            // 当前的 type 是非法的.
            throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
        }
        // 3. 构造响应
        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()
                + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

这一段逻辑看起来吓人，其实就是在处理请求中传递而来的 type ，根据不同 type 的类型来调用不同的方法

当然还有启动和关闭，这个就不用一步步分析了，大概来看看代码把：

这里还有关于连接没有讲到，等下一章继续完善最后的连接，

自定义协议响应代码

BrokerServer