浅谈如何自我实现一个消息队列服务器(7)——编写服务器部分

2024-05-10 19:12

本文主要是介绍浅谈如何自我实现一个消息队列服务器(7)——编写服务器部分,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 一、编写服务器代码
    • 1.1、分析一个服务器应具备的功能
      • 1.1.1、成员变量
      • 1.1.2、对外提供的接口

一、编写服务器代码

在这里插入图片描述
      再次拿出这张图,前面我们已经将重要概念:VirtualHost、exchange、msgQueue、message、binding 都实现了,此时就可以开始编写消息队列MQ的本体:BrokerServer (服务器),由于消息队列的服务器是一个基于 TCP 协议进行通信的服务器,因此消息队列的 BrokerServer 也叫做 TCP服务器。

1.1、分析一个服务器应具备的功能

首先分析一个服务器需要具备的功能来确定其类种应含有的字段及应对外提供的接口:

1.1.1、成员变量

1、ServerSocket
      服务器需要依靠 ServerSocket 进行网络通信来传输数据,因此需定义一个 ServerSocket类作为成员变量之一;

2、VirtualHost
      由于当前实现的MQ只支持一个虚拟主机(后面会继续完善该项目,使其能够支持多个虚拟主机,当前只是一个基础版本MQ),因此此时定义一个 VirtualHost 类作为 BrokerServer 类的成员变量之一;

3、ConcurrentHashMap<String,Socket> sessions = new ConcurrentHashMap<>();
      服务器会与多个客户端进行通信,使用哈希表来表示服务器当前与所有客户端的会话。
即当有客户端连接上服务器进行通信时,会讲客户端记录到当前哈希表,此处的 key 为 channelId(客户端与服务器建立连接,就会创建出一个TCP连接,该TCP连接就可以通过Cannel来包含多个逻辑上的子连接), Value 为 客户端连接 socket

4、ExecutorService
      一个服务器需要处理多个客户端请求,因此需要一个线程池来执行众多的客户端连接。

5、runnable
      需要一个布尔值来控制服务器的启动。true:服务器启动。正常工作。 false:服务器关闭,不进行工作。

1.1.2、对外提供的接口

1、服务器的构造方法:
      在该方法中,对 serverSocket 对象初始化,给此次服务器网络通信时绑定一个端口号。

2、服务器的启动方法:(public void start())
      2.1、给线程对象创建实例
      2.2、通过 accept() 获取客户端连接
      2.3、获取到客户端连接之后,将连接放入线程池中,由线程池中的线程处理。

3、处理一个客户端的连接的方法(一个TCP连接可以复用,因此一个TCP连接里可以有多个 channel,通过 channel ,一个TCP连接可以发出多个请求,收到多个响应):(public void processConnection(Socket clientSocket))
      3.1、读取 连接 中的流对象
      3.2、按照前面约定的应用层协议格式读取
      3.3、读取请求并解析(readRequest())
      3.4、根据请求计算出响应(process())
      3.5、服务器将响应写回客户端(writeResponse())
      3.6、连接处理完后需要关闭连接
      3.7、清除sessions里已经被关闭了连接的socket(clearClosedSocket())

4、服务器停止启动的方法:(public void stop())
      4.1、将 runnable = false、
      4.2、然后将线程池中要进行处理的所有连接全部销毁。
      4.3、关闭网络连接。

5、读取请求并解析的方法:(读取请求的方式跟当初约定的请求报文顺序一致。readRequest(DataInputStream dataInputStream))
      5.1、构造请求对象
      5.2、从流对象中读出4字节作为请求的 type,然后再将读出的 type 值设置到 request 对象中。
      5.3、从流对象中读出4字节作为请求的 length,然后再将读出的 length 值设置到 request 对象中。
      5.4、再从流对象中读出 payload 的长度。然后判断真实读到的 payload 长度 是否与 其 请求体长度 一致,不一致的话说明读的过程有问题,直接抛异常。
      5.5、返回请求

6、根据请求计算出响应的方法:(process(Request request,Socket clientSocket))
      6.1、先将 payload 反序列化,然后转成 BasicArguments 类型。
      6.2、将请求标识 rid 、连接标识 channelId、请求的 type、请求的 length 打印出来。方便后续判断当前是哪个连接,连接中的哪一对请求、响应。
       6.3、根据 type 值判断当前请求到底是需要进行什么操作??远程调用 API 的哪一个。
在这里插入图片描述
              6.3.1、不同的API其所具有的参数类都不同,不同的 type 值调用不同的参数类。
              6.3.2、再将virtualHost类里的对应方法的参数修改成当前request中携带的参数即可。
                          6.3.2.1、 注意:basicComsume()方法的第4个参数:回调函数,并不是客户端传过来的,而是服务器这边要有一个固定的简单、一致格式的回调函数,来把收到的消息回传给客户端(订阅者)。(那么首先服务器收到消息后,如何知道要将消息传给哪个客户端?通过回调函数里需要重写的 handleDelivery()方法里的参数 comsumerTag(消费者唯一身份标识),这时候 comsumerTag == channelId(客户端连接时的身份标识)。此时根据 channelId 就可以到 sessions 哈希表中查询到是哪个 Socket 对象,此时就可以往里面传消息了,此时客户端收到服务器传来的消息之后,就可以执行自己的回调函数,将消息消费掉。)
                                            6.3.2.1.1、在 sessions 中根据 comsumerTag 查找相应的客户端,如果客户端为null或已被关闭连接,此时订阅消息的客户端已经关闭,无法往里边发送消息了,直接抛异常。
                                            6.3.2.1.2、获取到socket对象且还在连接中,此时就可以构造 SubscribeReturns 对象、Response对象,设置好 SubscribeReturns 里的属性,设置好 response里的属性,payload 就是 SubscribeReturns 序列化后的结果。将响应写回客户端。
              6.3.3、赋值返回值ok。
       6.4、 整个 basicReturns 作为 response 的 payload。构造响应对象,设置好响应对象中的值,然后打印 rid、channelId、响应 type、响应 length。
       6.5、返回 response 对象。

7、将已经关闭了的客户端连接(socket)里的所有键值对都清理掉:(public void clearClosedSocket(Socket clientSocket))
       7.1、遍历哈希表中每个元素
       7.2、集合类中不能一边迭代一边删除,否则集合类的迭代器会由于结构被破坏而迭代失败,因此先把想删除的元素使用某数据结构存起来,然后再删除即可。
       7.3、打印提示信息。

这篇关于浅谈如何自我实现一个消息队列服务器(7)——编写服务器部分的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/977309

相关文章

Go语言开发实现查询IP信息的MCP服务器

《Go语言开发实现查询IP信息的MCP服务器》随着MCP的快速普及和广泛应用,MCP服务器也层出不穷,本文将详细介绍如何在Go语言中使用go-mcp库来开发一个查询IP信息的MCP... 目录前言mcp-ip-geo 服务器目录结构说明查询 IP 信息功能实现工具实现工具管理查询单个 IP 信息工具的实现服

Java的栈与队列实现代码解析

《Java的栈与队列实现代码解析》栈是常见的线性数据结构,栈的特点是以先进后出的形式,后进先出,先进后出,分为栈底和栈顶,栈应用于内存的分配,表达式求值,存储临时的数据和方法的调用等,本文给大家介绍J... 目录栈的概念(Stack)栈的实现代码队列(Queue)模拟实现队列(双链表实现)循环队列(循环数组

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结

springboot上传zip包并解压至服务器nginx目录方式

《springboot上传zip包并解压至服务器nginx目录方式》:本文主要介绍springboot上传zip包并解压至服务器nginx目录方式,具有很好的参考价值,希望对大家有所帮助,如有错误... 目录springboot上传zip包并解压至服务器nginx目录1.首先需要引入zip相关jar包2.然

将Java项目提交到云服务器的流程步骤

《将Java项目提交到云服务器的流程步骤》所谓将项目提交到云服务器即将你的项目打成一个jar包然后提交到云服务器即可,因此我们需要准备服务器环境为:Linux+JDK+MariDB(MySQL)+Gi... 目录1. 安装 jdk1.1 查看 jdk 版本1.2 下载 jdk2. 安装 mariadb(my

一文详解如何在Python中从字符串中提取部分内容

《一文详解如何在Python中从字符串中提取部分内容》:本文主要介绍如何在Python中从字符串中提取部分内容的相关资料,包括使用正则表达式、Pyparsing库、AST(抽象语法树)、字符串操作... 目录前言解决方案方法一:使用正则表达式方法二:使用 Pyparsing方法三:使用 AST方法四:使用字

SpringKafka错误处理(重试机制与死信队列)

《SpringKafka错误处理(重试机制与死信队列)》SpringKafka提供了全面的错误处理机制,通过灵活的重试策略和死信队列处理,下面就来介绍一下,具有一定的参考价值,感兴趣的可以了解一下... 目录引言一、Spring Kafka错误处理基础二、配置重试机制三、死信队列实现四、特定异常的处理策略五

在Android平台上实现消息推送功能

《在Android平台上实现消息推送功能》随着移动互联网应用的飞速发展,消息推送已成为移动应用中不可或缺的功能,在Android平台上,实现消息推送涉及到服务端的消息发送、客户端的消息接收、通知渠道(... 目录一、项目概述二、相关知识介绍2.1 消息推送的基本原理2.2 Firebase Cloud Me

浅谈配置MMCV环境,解决报错,版本不匹配问题

《浅谈配置MMCV环境,解决报错,版本不匹配问题》:本文主要介绍浅谈配置MMCV环境,解决报错,版本不匹配问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录配置MMCV环境,解决报错,版本不匹配错误示例正确示例总结配置MMCV环境,解决报错,版本不匹配在col

基于Python打造一个可视化FTP服务器

《基于Python打造一个可视化FTP服务器》在日常办公和团队协作中,文件共享是一个不可或缺的需求,所以本文将使用Python+Tkinter+pyftpdlib开发一款可视化FTP服务器,有需要的小... 目录1. 概述2. 功能介绍3. 如何使用4. 代码解析5. 运行效果6.相关源码7. 总结与展望1