项目实战 — 消息队列（8）{网络通信设计①}

news/2024/11/7 22:35:10/

目录

一、自定义应用层协议

🍅 1、格式定义

🍅 2、准备工作

🎄定义请求和响应 

 🎄 定义BasicArguments

🎄 定义BasicReturns

🍅 2、创建参数类

        🎄 交换机

        🎄 队列

        🎄 绑定

        🎄发布消息

        🎄 订阅消息

        🎄确认应答

        🎄 消息推送

二、服务器设计

 🍅 1、编写实例变量和构造方法

🍅 2、编写启动类和关闭类

🍅 3、编写处理连接的方法：processConnection()

 🍅 4、编写读取请求readRequest()和写回响应writeResponse方法

🍅 5、实现根据请求计算响应：process()方法编写

一、自定义应用层协议

🍅 1、格式定义

本消息队列，是需要通过网络进行通信的。这里主要基于TCP协议，自定义应用层协议。

由于当前交互的Message数据，是二进制数据，由于HTTP和JSON都是文本协议，所以这里就不适用了。使用自定义的应用层协议。

约定自定义应用层协议格式:

        以下是请求和响应的组成部分：

 type：

描述当前请求和响应式做什么的，描述当前请求/响应是在调用哪个API（VirtualHost中的核心API）

        以下是type标识请求相应不同的功能，取值如下：

        其中Channel代表的是Connection（TCP的连接）内部的”逻辑上"的连接。此时一个           Connection中可能会含有多个Channel。存在的意义是为了让TCP连接

VirtualHost中的十多个方法：
0x1创建channel
0x2关闭channel
0x3创建exchange
0x4销毁exchange
0x5创建queue
0x6销毁queue
0x7创建binding
0x8销毁binding
0x9发送message
0xa订阅message
0xb返回ack
0xc服务器给客户端推送的消息（被订阅的消息）（响应独有）

length：描述了payload的长度

payload： 会根据当前是请求还是响应，以及当前的type有不同的取值。

比如当前是0x3（创建交换机），

/*
* 表示一个网络通信中的请求对象，按照自定义协议的格式来展开
* */
@Data
public class Request {private int type;private int length;private byte[] payload;
}

当前是一个请求，那么pyload中的内容是exchangeDeclare的参数的序列化的结果；

如果当前是一个响应，那么payload里面的内容就是exchangeDeclare的返回结果的序列化内容。

那么接下来就进行代码设计

以下都是再commen包中创建。

🍅 2、准备工作

🎄定义请求和响应 

/*
* 表示一个网络通信中的请求对象，按照自定义协议的格式来展开
* */
@Data
public class Request {private int type;private int length;private byte[] payload;
}
/*
* 表示一个网络通信中的响应对象,也是根据自定义应用层协议来的
* */
@Data
public class Response {private int type;private int length;private byte[] payload;
}

 🎄 定义BasicArguments

使用这个类表示方法的公共参数/辅助的字段 ，后续的每个方法会有一些不同的参数,不同的参数再使用不同的子类来表示。

rid代表请求的id，和响应的id一样，他们是一对

channel表示的是“逻辑连接”，表示客户端各种模块复用一个TCP连接，

channelId就代表这些连接。

@Data
public class BasicArguments implements Serializable {
//     表示一次请求/响应的身份标识，可以把请求和响应对上protected String rid;
//    客户端的身份标识protected String channelId;
}

🎄 定义BasicReturns

使用这个类标识各个远程调用的方法的返回值的公共信息

/*
* 标识各个远程调用的方法的返回值的公共信息
* */
@Data
public class BasicReturns implements Serializable {
//    用来标识唯一的请求和响应protected String rid;protected String channelId;
//    用来表示当前远程调用方法的返回值protected boolean ok;
}

🍅 2、创建参数类

根据前面VirtualHost中的十多个方法，每个方法创建一个类，标识该方法中的相关参数。

那么这个参数到底是如何进行传递的？

如下图，以交换机的参数进行举例。

关于我们远程调用的过程：当发起请求时，就把这些参数通过请求传过去，然后调用VirtualHost中的API（就是VirtualHost中的那些创建删除方法），调用完以后再返回响应。

以下是有关交换机的请求报文：

以下是创建交换机的响应报文：没有请求报文复杂是因为，响应只需要返回请求是否执行远程调用是否成功即可。 

以下就创建这些参数类： 

        🎄 交换机

 创建交换机：

@Data
public class ExchangeDeclareArguments extends BasicArguments implements Serializable {private String ExchangeName;private ExchangeType exchangeType;private boolean durable;
}

删除交换机：

@Data
public class ExchangeDeleteArguments extends BasicArguments implements Serializable {private String exchangeName;
}

        🎄 队列

创建队列：

@Data
public class QueueDeclareArguments extends BasicArguments implements Serializable {private String QueueName;private boolean durable;
}

删除队列：

@Data
public class QueueDeleteArguments extends BasicArguments implements Serializable {private String queueName;
}

        🎄 绑定

创建绑定：

@Data
public class QueueBindArguments extends BasicArguments implements Serializable {private String exchangeName;private String queueName;private String bindingKey;
}

删除绑定：

@Data
public class QueueUnbindArguments extends BasicArguments implements Serializable {private String queueName;private String exchangeName;
}

        🎄发布消息

@Data
public class BasicPublishArguments extends BasicArguments implements Serializable {private String exchangeName;private String routingKey;private BasicProperties basicProperties;private byte[] body;
}

        🎄 订阅消息

这个方法参数，还包含一个Consumer consumer。

这是一个回调函数，这个回调函数是不能作为参数进行传输的，因为这个回调函数，是客户端这边的。

比如，这里请求调用一个”订阅队列“的远程方法，

客户端这边：服务器收到了请求，执行了basicConsume方法，并且返回了响应。订阅以后，客户端的消费者就会在后面收到消息，而这个回调函数是在消费者收到消息以后，才会进行逻辑处理，而不是再发送请求时进行传递的。

服务器这边：执行的是一个固定的回调函数：把消息返回给客户端。

@Data
public class BasicConsumeArguments extends BasicArguments implements Serializable {private String consumerTag;private String queueName;private boolean autoAck;
}

        🎄确认应答

@Data
public class BasicAckArguments extends BasicArguments implements Serializable {private String queueName;private String messageId;
}

        🎄 消息推送

前面的都是客户端给服务器发送消息，这里是服务器给消费者推送消息。所以要继承BasicReturns。

@Data
public class SubScribeReturns extends BasicReturns implements Serializable {private String consumerTag;private BasicProperties basicProperties;private byte[] body;
}

二、服务器设计

在 mqServer包中创建一个BrokerServer类。

 🍅 1、编写实例变量和构造方法

 private ServerSocket serverSocket = null;private VirtualHost virtualHost = new VirtualHost("default");//    使用这个哈希表，表示当前所有会话（那些客户端在和这个服务器进行通信）
//    此处的key是channelId，value是对应的 socket对象private ConcurrentHashMap<String , Socket> sessions = new ConcurrentHashMap<String ,Socket>();//    引入线程池，处理多个客户端的请求private ExecutorService executorService = null;//    引入boolean变量控制服务器是否运行private volatile boolean runnable = true;public BrokerServer(int port) throws IOException {
//        端口号serverSocket = new ServerSocket(port);}

🍅 2、编写启动类和关闭类

 这里利用了线程池，不断的处理连接

    public void start() throws IOException {System.out.println("[BrokerServer]启动");
//        定义一个线程池。处理客户端的连接请求executorService = Executors.newCachedThreadPool();while (runnable){Socket clientSocket = serverSocket.accept();
//            把处理连接的逻辑给线程池executorService.submit(()->{try {processConnection(clientSocket);} catch (IOException e) {throw new RuntimeException(e);}});}}public void stop() throws IOException {runnable = false;
//        停止线程池executorService.shutdownNow();serverSocket.close();}private void processConnection(Socket clientSocket) {//TODO
}

🍅 3、编写处理连接的方法：processConnection()

处理一个客户端的连接，主要有以下几步：

        （1）读取请求并且解析

        （2）根据请求计算响应

        （3）把相应协写回给客户端

//    通过该方法，处理一个客户端的连接
//    在一个连接中，可能会涉及到多个连接和请求private void processConnection(Socket clientSocket) throws IOException {
//        获取到流对象，读取应用层协议try(InputStream inputStream = clientSocket.getInputStream();OutputStream outputStream = clientSocket.getOutputStream()){
//                按照特定格式来读取并且解析（转换），此时就需要用到DataInputStream和DataOutputStreamtry(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) {
//                DataInputStream如果读到EOF（文件末尾），会抛出一个EOFException异常
//                视为正常的异常，用或者异常来结束循环System.out.println("[BrokerServer]connection关闭！客户端的地址：" + clientSocket.getInetAddress().toString()+ ":" + clientSocket.getPort());}catch (IOException | ClassNotFoundException | MqException e){
//            不正常的异常System.out.println("[BrokerServer]connection出现异常");e.printStackTrace();}finally {clientSocket.close();
//          一个TCP连接中，可能含有多个channel,需要把当前socket对应的channel也顺便清理掉clearClosedSession(clientSocket);}}

 🍅 4、编写读取请求readRequest()和写回响应writeResponse方法

这里就是根据前面设定的报文格式来编写的读取请求和写回响应的方法，这里的payload的具体内容在这里不作解析，在后面的process方法中进行解析

//    读取请求并且解析private Request readRequest(DataInputStream dataInputStream) throws IOException {Request request = new Request();//        读取出请求中4个字节的typerequest.setType(dataInputStream.readInt());
//        读出4个字节的lengthrequest.setLength(dataInputStream.readInt());byte[] payload = new byte[request.getLength()];int n = dataInputStream.read(payload);if (n != request.getLength()){throw new IOException("读取请求格式出错");}request.setPayload(request.getPayload());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();}

🍅 5、实现根据请求计算响应：process()方法编写

这里就要针对具体的payload进行编写了。

当前请求中的payload里面的内容，是根据type来的，如下

VirtualHost中的十多个方法：
0x1创建channel
0x2关闭channel
0x3创建exchange
0x4销毁exchange
0x5创建queue
0x6销毁queue
0x7创建binding
0x8销毁binding
0x9发送message
0xa订阅message
0xb返回ack
0xc服务器给客户端推送的消息（被订阅的消息）（响应独有）

如果是0x3，就是创建交换机对应的参数...... 

主要分为以下几步：

        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){
//            创建channelsessions.put(basicArguments.getChannelId(),clientSocket);System.out.println("[BrokerServer]创建channel完成！channelId = " + basicArguments.getChannelId());}else if (request.getType() == 0x2){sessions.remove(basicArguments.getChannelId());System.out.println("[BrokerServer]销毁完成！channelId = " + basicArguments.getChannelId());}else if (request.getType() == 0x3){
//            创建交换机ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;ok = virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getExchangeType(),arguments.isDurable());}else if (request.getType() == 0x4){ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;ok = virtualHost.exchangeDelete(arguments.getExchangeName());}else if (request.getType() == 0x5){QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable());}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() {
//                            该回调函数，将服务器收到的消息推送回对应的消费者客户端@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
//                            先知道当前收到的消息，要发送给那个客户端，
//                            此处的consumerTag其实就是channelId,根据channelId去session中查询,就可以得到对应的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);
//                            此处rid不设置，因为这里只有响应没有请求，rid不需要去对应subScribeReturns.setRid("");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);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 +",type" + response.getType() + ",length = " + response.getLength());return response;}

🍅 6、清理过期的sessions：clearClosedSession()

    //    遍历sessions hash表，把该被关闭的socket对应的键值对都删掉private void clearClosedSession(Socket clientSocket) {List<String> toDeleteChannelId = new ArrayList<>();for(Map.Entry<String,Socket> entry : sessions.entrySet()){if(entry.getValue() == clientSocket){
//                使用集合类，不能一边遍历，一边删除toDeleteChannelId.add(entry.getKey());}}for (String channelId : toDeleteChannelId){sessions.remove(channelId);}System.out.println("[BrokerServer]清理session完成~ 被清理的channeId = " + toDeleteChannelId);}


http://www.ppmy.cn/news/1027862.html

相关文章

2023河南萌新联赛第（五）场：郑州轻工业大学C-数位dp

2023河南萌新联赛第（五）场：郑州轻工业大学C-数位dp

链接&#xff1a;登录—专业IT笔试面试备考平台_牛客网 给定一个正整数 n&#xff0c;你可以对 n 进行任意次&#xff08;包括零次&#xff09;如下操作&#xff1a; 选择 n 上的某一数位&#xff0c;将其删去&#xff0c;剩下的左右部分合并。例如 123&#xff0c;你可以选择…
阅读更多...
MATLAB 2023a的机器学习、深度学习

MATLAB 2023a的机器学习、深度学习

MATLAB 2023版的深度学习工具箱&#xff0c;提供了完整的工具链&#xff0c;使您能够在一个集成的环境中进行深度学习的建模、训练和部署。与Python相比&#xff0c;MATLAB的语法简洁、易于上手&#xff0c;无需繁琐的配置和安装&#xff0c;让您能够更快地实现深度学习的任务。…
阅读更多...
【linux】2 软件管理器yum和编辑器vim

【linux】2 软件管理器yum和编辑器vim

目录 1. linux软件包管理器yum 1.1 什么是软件包 1.2 关于rzsz 1.3 注意事项 1.4 查看软件包 1.5 如何安装、卸载软件 1.6 centos 7设置成国内yum源 2. linux开发工具-Linux编辑器-vim使用 2.1 vim的基本概念 2.2 vim的基本操作 2.3 vim正常模式命令集 2.4 vim末行…
阅读更多...
FileZilla Server安装配置使用说明

FileZilla Server安装配置使用说明

作者&#xff1a;John 链接&#xff1a;https://www.zhihu.com/question/20577011/answer/2360828234 来源&#xff1a;知乎 第一步&#xff1a;右键点击”立即下载“ 第二步&#xff1a;服务器端点击&#xff0c;“windows平台”版本 第三步&#xff1a;这个安装最新的“Fi…
阅读更多...
Django中级指南：理解并实现Django的模型和数据库迁移

Django中级指南：理解并实现Django的模型和数据库迁移

Django 是一个极其强大的 Python Web 框架&#xff0c;它提供了许多工具和特性&#xff0c;能够帮助我们更快速、更便捷地构建 Web 应用。在本文中&#xff0c;我们将会关注 Django 中的模型&#xff08;Models&#xff09;和数据库迁移&#xff08;Database Migrations&#x…
阅读更多...
详细介绍如何对音乐信息进行检索和音频节拍跟踪

详细介绍如何对音乐信息进行检索和音频节拍跟踪

在本文中，我们将了解节拍的概念，以及我们在尝试跟踪节拍时面临的挑战。然后我们将介绍解决问题的方法以及业界最先进的解决方案。 介绍 音乐就在我们身边。每当我们听到任何与我们的心灵和思想相关的音乐时，我们就会迷失其中。我们下意识地随着听到的节拍而敲击。您一定已…
阅读更多...
C++unique_ptr小结

C++unique_ptr小结

文章目录 返回unique_ptr指定删除器尺寸问题 返回unique_ptr unique_ptr<string> test() {return unique_ptr<string>(new string("hello world"));// 我们知道在unique_ptr不能进行赋值, 或者使用其他的指针初始化, // 在这里创建了临时的对象, 所以会…
阅读更多...
Titanic--细节记录二

Titanic--细节记录二

merge、join以及concat的方法的不同以及相同 相同之处&#xff1a;都用于合并数据。 不同之处&#xff1a; merge主要是基于列的合并。join主要是基于索引&#xff08;行标签&#xff09;的合并。concat可以沿任意轴合并&#xff0c;更灵活。 import pandas as pddf1 pd.Da…
阅读更多...
最新文章

Copyright @ 2022~2023