浅谈如何自我实现一个消息队列服务器(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、打印提示信息。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/687286.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【AI+换脸换装】从OpenAI 探索色情露骨内容领域浅聊AI换脸换装

5月9日消息&#xff0c;据外电报道&#xff0c;OpenAI 周三发布了文档草案&#xff0c;阐述了它希望 ChatGPT 及其其他人工智能技术如何运作。冗长的Model Spec 文件的一部分透露&#xff0c;该公司正在探索进军色情和其他露骨内容领域。 看完这个&#xff0c;心里有点惊讶&am…

以蒸馏的名义:“从去噪自编码器到生成模型”重出江湖

©PaperWeekly 原创 作者 | 苏剑林 单位 | 科学空间 研究方向 | NLP、神经网络 今天我们分享一下论文《Score identity Distillation: Exponentially Fast Distillation of Pretrained Diffusion Models for One-Step Generation》[1]&#xff0c;顾名思义&#xff0c;这是…

【HMGD】GD32/STM32 DMA接收不定长串口数据

单片机型号&#xff1a;GD32F303系列 CubeMX配置 配置串口参数 开启DMA 开启中断 示例代码 使用到的变量 uint8_t RX_Buff_FLAG 0; uint8_t RX_Buff[300] {0}; uint8_t TX_Buff[300] {0};串口接收空闲函数 // 串口接收空闲函数 void HAL_UARTEx_RxEventCallback(UART_H…

初识C语言——第十七天

选择语句&#xff1a;switch switch语句&#xff08;整型表达式&#xff09; { 语句项&#xff1a; } 而语句项是什么呢&#xff1f; //是一些case语句&#xff1a; //如下 case 整形常量表达式&#xff1b;常量可以&#xff0c;字符也可以&#xff08;因为字符存储的时…

灯珠CCD或CMOS成像RGB数据 光谱重建

1. 源由 本文主要为了通过摄像头CCD或者CMOS传感器对灯珠成像数据分析、重建灯珠可见光范围光谱数据的研究&#xff0c;从原理和方法上论证可行性。 随着照明技术迅猛发展&#xff0c;LED技术日渐成熟。LED产品由于具备经久耐用、节能且价格低等优势&#xff0c;已成为照明行…

算法学习011-不同的二叉查找树/搜索树 c++动态规划算法实现 中小学算法思维学习 信奥算法解析

目录 C不同的二叉查找树 一、题目要求 1、编程实现 2、输入输出 二、算法分析 三、程序编写 四、运行结果 五、考点分析 六、推荐资料 C不同的二叉查找树 一、题目要求 1、编程实现 二叉查找树&#xff08;Binary Search Tree&#xff09;&#xff0c;&#xff08;…

STC8增强型单片机开发【定时器Timer⭐】

目录 一、引言 二、定时器基础知识 三、STC8定时器配置 四、代码示例 五、总结 一、引言 在单片机开发中&#xff0c;定时器&#xff08;Timer&#xff09;是一个极其重要的组件&#xff0c;它允许开发者基于时间触发各种事件或任务。STC8增强型单片机作为一款功能丰富的…

uni-app(三):离线打包与插件引用(Android)

离线打包与插件引用 1.下载Android离线SDK2.使用Android Studio打开离线打包项目并更新Gradle3.解决报错4.构建5.配置AppKeya.查看证书b.申请AppKeyc.配置AppKey 6.生成本地打包App资源7.拷贝App资源到Android项目中8.修改 appid9.修改Android项目配置文件10.下载证书并配置11.…

Sqli-labs第五~八关(布尔盲注)

目录 首先找到他们的闭合方式 操作 总结&#xff1a; 第五关根据页面结果得知是字符型但是和前面四关还是不一样是因为页面虽然有东西。但是只有对于请求对错出现不一样页面其余的就没有了。这个时候我们用联合注入就没有用&#xff0c;因为联合注入是需要页面有回显位。如果…

day07beef-xss之根据beef-xss获取cookies

1.安装 apt-get update apt-get install beef-xss 若报错运行不了尝试 apt remove ruby apt remove beef-xss apt-get install ruby apt-get install ruby-dev libpcap-dev gem install eventmachine apt-get install beef-xss 2.运行 beef-xss 运行成功会自动弹出浏览框。 攻…

Web实时通信的学习之旅:轮询、WebSocket、SSE的区别以及优缺点

文章目录 一、通信机制1、轮询1.1、短轮询1.2、长轮询 2、Websocket3、Server-Sent Events 二、区别1、连接方式2、协议3、兼容性4、安全性5、优缺点5.1、WebSocket 的优点&#xff1a;5.2、WebSocket 的缺点&#xff1a;5.3、SSE 的优点&#xff1a;5.4、SSE 的缺点&#xff1…

AI大模型探索之路-训练篇20:大语言模型预训练-常见微调技术对比

系列篇章&#x1f4a5; AI大模型探索之路-训练篇1&#xff1a;大语言模型微调基础认知 AI大模型探索之路-训练篇2&#xff1a;大语言模型预训练基础认知 AI大模型探索之路-训练篇3&#xff1a;大语言模型全景解读 AI大模型探索之路-训练篇4&#xff1a;大语言模型训练数据集概…