紧接着上一章没说完的进行服务器的补充。
推送给消费者消息的基本实现思路
- 让 brokerServer 把哪些消费者管理好
 - 收到对应的消息,把消息推送给消费者
 
消费者是以队列为维度来订阅消息的,一个队列可以有多个消费者(此处我们约定按照轮询的方式来进行消费)

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

这里又一次提到了消费者,我们来把消费者相关的代码完善一下。
消费者管理实现
消费者
先前我们提到了这个函数式接口,这个接口的作用就是用来消费消息(和消费者作用一样)。
那么我们就给这个函数式接口起名消费者 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);
            }
        }
    } 
消费消息
大致逻辑
- 按照轮询的方法,找个消费者出来(没有消费者就暂时不消费,等有消费者出现才进行消费)
 - 从队列中取出一个消息(当前队列中还没有消息,也不需要消费)
 - 把消息带入到消费者的回调方法中, 丢给线程池执行 
  
- 把消息放到待确认的集合中. 这个操作势必在执行回调之前
 - 真正执行回调操作
 - 如果当前是 "自动应答" , 就可以直接把消息删除了. 如果当前是 "手动应答" , 则先不处理, 交给后续消费者调用 basicAck 方法来处理. 
    
- 删除硬盘上的消息
 - 删除上面的待确认集合中的消息
 - 删除内存中消息中心里的消息
 
 
 
具体代码如下:
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)
这一步属于重中之重。
大致流程如下:
- 把 request 中的 payload 做一个初步解析
 - 根据 type 的值,决定具体要干啥
 - 构造响应
 
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



















