模拟实现消息队列项目(系列7) -- 实现BrokerServer

目录

前言

1. 创建BrokerServer类

1.1 启动服务器

1.2 停止服务器

1.3 处理一个客户端的连接

1.3.1 解析请求得到Request对象

1.3.2 根据请求计算响应

1.3.3 将响应写回给客户端

1.3.4 遍历Session的哈希表,把断开的Socket对象的键值对进行删除

2. 处理订阅消息请求详解(补充)

3. 序列化/反序列化实现(补充)

结语

---

前言

        上一章节,我们定义了本项目的应用层传输协议.并且创建了各种参数类.本章节的目标是对BrokerServer(实现一个TCP服务器)进行实现,对连接进行处理,根据请求计算响应返回给客户端.

---

1. 创建BrokerServer类

public class BrokerServer {// 当前考虑一个一个服务器中只有一个虚拟主机private VirtualHost virtualHost = new VirtualHost("default");// 使用哈希表表示当前会话,也就是有哪些客户端在和服务器进行通信// key: channelId value:对应对的Socket对象private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();private ServerSocket serverSocket = null;// 引入线程池来处理多个客户端private ExecutorService executorService = null;// 控制服务器是否继续运行private volatile boolean runnable = true;
}

1.1 启动服务器

1. 首先将线程池进行创建,用来处理多个连接.

2. 设置循环用来监听连接

3. 将处理连接交给线程池.

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

1.2 停止服务器

1. 将标志位runnable设置为false

2. 停止线程池的服务

3. 关闭服务器套接字

/*** 2. 停止服务器*/public void stop() throws IOException {runnable = false;// 停止线程池executorService.shutdownNow();serverSocket.close();}

1.3 处理一个客户端的连接

1. 我们是从请求中获取的信息是二进制文件,我们不能直接使用InputStream和OutputStream,我们借助DataInputStream和DataOutputStream进行操作字节流.

2. 使用DataInputStream进行读取请求的时候,读到末尾的时候会抛出一个异常,我们将这个异常视作为处理正确的业务逻辑.我么catch掉这个异常就可以.

3. 解析得到请求对象

4. 更具请求计算响应

5. 当处理完响应之后,要进行关闭连接,并且将一个连接中其他Channel进行关闭.

/*** 3. 处理一个客户端的连接*    在一个连接中会出现多个请求和多个响应.在一个连接中要循环的处理*/private void processConnection(Socket clientSocket) {try(InputStream inputStream = clientSocket.getInputStream();OutputStream outputStream = clientSocket.getOutputStream()){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(dataOutputStream, response);}}}catch (EOFException | SocketException e){// 处理正确的业务逻辑// 上述进行读取数据的时候,如果数据读到末尾(EOF) ,就会抛出一个异常// 借助这个异常结束上述循环System.out.println("[BrokerServer]  连接关闭! 客户端地址:" +clientSocket.getInetAddress().toString()+ ",端口号: "+clientSocket.getPort());}catch (ClassNotFoundException | MqException e) {e.printStackTrace();} catch (IOException e) {// 处理真正的异常System.out.println("[BrokerServer] connection 出现异常");e.printStackTrace();}finally {try {// 当连接处理完成之后,进行关闭连接clientSocket.close();// 一个连接中可能会包含多个channel,需要把当前这个Socket对应的所有channel进行关闭clearClosedSession(clientSocket);} catch (IOException e) {e.printStackTrace();}}}

1.3.1 解析请求得到Request对象

1. 根据我们自定义的格式,先读前4个字节是请求的类型,在读4个字节是payload的长度,在读就是payload.

2. 读取payload的时候,我们先根据长度创建字符数组,然后按照字符数组进行获取payload,比较读取完的长度是否与原来请求的长度一致,不一致说明有消息的丢失.进行抛出异常.

3. 最后得到完整的请求对象,交给下面的方法进行处理.

 /*** 3.1 解析请求得到Request对象*/private Request readRequest(DataInputStream dataInputStream) throws IOException {Request request = new Request();// 1. 首先读取四个字节,为请求的typerequest.setType(dataInputStream.readInt());// 2. 在读四个字节就是payload的长度request.setLength(dataInputStream.readInt());// 3. 创建字符数组,并进行读取到数组中byte[] payload = new byte[request.getLength()];int n = dataInputStream.read(payload);if (n != request.getLength()){throw new IOException("[BrokerServer] 请求格式出错");}// 4. 将读取的数组内容写入到实体的Request对象中request.setPayload(payload);return request;}

1.3.2 根据请求计算响应

1. 我们根据请求对象的payload进行解析,此处需要注意的是,我们读取到的payload是字节数组,我们需要进行反序列化成字符数组.

2. 根据请求对象的Type值进行区分,到底客户端要调用服务器那些功能.

3. 处理完请求之后就要进行构造响应了.

4. 返回响应对象

/*** 3.2 根据请求计算响应* @param request* @param clientSocket* @return*/private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {// 1. 根据request中的payload进行解析//  payload 是根据 request 中 type 进行变化的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){// 1. 创建一个channelsessions.put(basicArguments.getChannelId(),clientSocket);System.out.println("[BrokerServer] 创建channel完成 getChannelId="+ basicArguments.getChannelId());}else if (request.getType() == 0x2){// 2. 销毁一个channelsessions.remove(basicArguments.getChannelId());System.out.println("[BrokerServer] 销毁channel完成 getChannelId="+ basicArguments.getChannelId());}else if (request.getType() == 0x3){// 3. 创建交换机ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;// 调用虚拟主机的功能方法ok = virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getExchangeType(),arguments.isDurable(),arguments.isAutoDelete(),arguments.getArguments());System.out.println("[BrokerServer] 创建交换机完成 ExchangeName="+ arguments.getExchangeName());}else if (request.getType() == 0x4){// 4. 销毁交换机ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;ok = virtualHost.exchangeDelete(arguments.getExchangeName());System.out.println("[BrokerServer] 删除交换机完成 ExchangeName="+ arguments.getExchangeName());}else if (request.getType() == 0x5) {// 5. 创建队列QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(),arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());System.out.println("[BrokerServer] 创建队列完成 QueueName="+ arguments.getQueueName());} else if (request.getType() == 0x6) {// 6. 删除队列QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;ok = virtualHost.queueDelete((arguments.getQueueName()));System.out.println("[BrokerServer] 删除队列完成 QueueName="+ arguments.getQueueName());} else if (request.getType() == 0x7) {// 7. 创建绑定QueueBindArguments arguments = (QueueBindArguments) basicArguments;ok = virtualHost.queueBind(arguments.getQueueName(),arguments.getExchangeName(), arguments.getBindingKey());System.out.println("[BrokerServer] 创建绑定完成 QueueName="+ arguments.getQueueName()+ ",ExchangeName=" + arguments.getExchangeName());} else if (request.getType() == 0x8) {// 8. 删除绑定QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());System.out.println("[BrokerServer] 删除绑定完成 QueueName="+ arguments.getQueueName()+ ",ExchangeName=" + arguments.getExchangeName());} else if (request.getType() == 0x9) {// 9. 发布消息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) {// 10. 订阅消息BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;ok = virtualHost.basicConsume(arguments.getConsumeTag(), arguments.getQueueName(), arguments.isAutoAck(), new Consumer() {@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws IOException, MqException {// 将服务器收到的消息进行推送给客户端// 先知道当前这个收到的消息, 要发给哪个客户端.// 此处 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 就是一个 SubScribeReturnsresponse.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) {// 10. 调用 basicAck 确认消息.BasicAckArguments arguments = (BasicAckArguments) basicArguments;ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());System.out.println("[BrokerServer] 消费者确认消息完成 QueueName=" + arguments.getQueueName()+ ", MessageId=" + 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;}

1.3.3 将响应写回给客户端

1. 注意写入类型和长度是写入固定的4个字节,那么我们就使用dataOutputStream.writeInt()

2. 写完响应之后,记得要刷新缓冲区 dataOutputStream.flush();

/*** 3.3 将响应写回给客户端* @param dataOutputStream* @param response*/private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {// 将响应的属性从计算好的响应中进行设置dataOutputStream.writeInt(response.getType());dataOutputStream.writeInt(response.getLength());dataOutputStream.write(response.getPayload());// 刷新缓冲区dataOutputStream.flush();}

1.3.4 遍历Session的哈希表,把断开的Socket对象的键值对进行删除

由于Socket都已经断开连接了,那么存储在内存中的Session也就没有存在的必要了.这个集合中存放的是一个连接中的change对应的Session,当连接断开之后,Channel也就不会再进行工作了,新的连接会创建新的Channel.

注意:我们在使用Map.entrySet进行遍历Map的时候,不要一遍遍历一遍进行删除,这样是不稳定的,我们遍历Map将需要进行移除的Session进行添加到待删除的链表中,最后遍历待删除的数据结构进行删除.

---

上述就是整个封装好的BrokerServer服务器.

---

下面呢,我对有关根据请求计算响应中订阅消息这一功能,再进行详细的阐述,这块比较难以理解,因为涉及到回调函数,大家可能不知道这个回调函数掉用的时机是哪里.

2. 处理订阅消息请求详解(补充)

 第二个红框部分是回调函数.

        只有消费者订阅的队列中有消息了,并且轮询的方式选中了这个消费者,才会获得消息的本体,此时线程池才会执行到这个回调方法,此时才拿到消息的本体,可以将消息的属性和本体写入到SubscribeReturn中,进而推送给消费者进行消费消息.如果没有消息给这个消费者,那么也不会进行断开连接,只要服务器不断开连接客户端一直在等待分配的消息进行消费.这一点希望,读者能够进一步的理解.等总结完客户端,那么我就会带着大家,再来理一遍这个订阅消息的这个思路.

3. 序列化/反序列化实现(补充)

要想能进行序列化和反序列化就必须对目标对象进行实现serializable接口.

1. 我们使用ByteArrayOutputStream和ObjectOutputStream进行将一个对象序列化为字节数组(输出的是字节用output)

2. 我们ByteArrayInputStream和ObjectInputStream将一个字节数组反序列化成一个对象(输入的是字节用Input)

---

结语

        至此,我们就彻底的完成了mqserver 的搭建,只剩下mqclient的搭建,我们在下一系列完成客户端的搭建,请持续关注,谢谢!!!

        完整的项目代码已上传Gitee,欢迎大家访问.👇👇👇

模拟实现消息队列https://gitee.com/yao-fa/advanced-java-ee/tree/master/My-mq