• 2024.2.27 模拟实现 RabbitMQ —— 网络通信设计（客户端）

    目录

    需求分析

    RabbitMQ 客户端设定

    ConnectionFactory（连接工厂）

    Connection（连接）

    Channel（通道）

    针对 客户端 和 服务器 单元测试

    需求分析

    RabbitMQ 客户端设定

    • 一个客户端可以有多个模块
    • 每个模块均可以和 Broker Server 之间建立 "逻辑上的连接"（channel）
    • 这几个模块的 channel 彼此之间是相互不影响的
    • 同时这几个 channel 复用了同一个 TCP 连接
    • 此处我们将仿照 RabbitMQ 客户端设定

    ConnectionFactory（连接工厂）

    • 这个类持有服务器的地址
    • 该类用于创建 Connection 对象

    具体代码编写：

    1. import lombok.Getter;
    2. import lombok.Setter;
    3.  
    4. import java.io.IOException;
    5.  
    6. @Getter
    7. @Setter
    8. public class ConnectionFactory {
    9. //    broker server 的 ip 地址
    10.     private String host;
    11. //    broker server 的端口号
    12.     private int port;
    13.  
    14.     public Connection newConnection() throws IOException {
    15.         Connection connection = new Connection(host,port);
    16.         return connection;
    17.     }
    18.  
    19. //    访问 broker server 的哪个虚拟主机
    20. //    下列几个属性暂时先不搞了
    21. //    private String virtualHostName;
    22. //    private String username;
    23. //    private String password;
    24. }

    Connection（连接）

    • 这个类表示一个 TCP 连接，持有 Socket 对象
    • 该类用于写入请求/读取响应，管理多个 Channel 对象

    具体代码编写：

    • 编写成员变量
    1.     private Socket socket = null;
    2. //    需要管理多个 channel 使用一个 hash 表把若干个 channel 组织起来
    3.     private ConcurrentHashMap channelMap = new ConcurrentHashMap<>();
    4.  
    5.     private InputStream inputStream;
    6.     private OutputStream outputStream;
    7.     private DataOutputStream dataOutputStream;
    8.     private DataInputStream dataInputStream;
    9.  
    10. //用来处理 0xc 的回调，这里开销可能会很大，不希望把 扫描线程 阻塞住，因此使用 线程池 来处理
    11.     private ExecutorService callbackPool = null;
    • 编写构造方法
    • 此处不仅需要初始化成员变量，还需创建一个扫描线程，不停的从 socket 中读取响应数据，并将读取到的响应交给 dispatchResponse 方法执行
    1.  public Connection(String host, int port) throws IOException {
    2.         socket = new Socket(host,port);
    3.         inputStream = socket.getInputStream();
    4.         outputStream = socket.getOutputStream();
    5.         dataInputStream = new DataInputStream(inputStream);
    6.         dataOutputStream = new DataOutputStream(outputStream);
    7.  
    8.         callbackPool = Executors.newFixedThreadPool(4);
    9.  
    10. //        创建一个 扫描线程，由这个线程负责不停的从 socket 中取响应数据 把这个响应数据再交给对应的 channel 负责处理
    11.         Thread t = new Thread(() -> {
    12.             try {
    13.                 while (!socket.isClosed()) {
    14.                     Response response = readResponse();
    15.                     dispatchResponse(response);
    16.                 }
    17.             }catch (SocketException e) {
    18. //                连接正常断开，此时这个异常直接忽略
    19.                 System.out.println("[Connection] 连接正常断开！");
    20.             }catch (IOException | ClassNotFoundException | MqException e) {
    21.                 System.out.println("[Connection] 连接异常断开！");
    22.                 e.printStackTrace();
    23.             }
    24.         });
    25.         t.start();
    26.     }
    • 编写 dispatchResponse 方法
    • 使用该方法来区分，当前响应是一个针对控制请求的响应，还是服务器推送过来的消息
    • 如果是服务器推送过来的消息，type = 0xc，也就需要执行回调，通过线程池来执行
    • 如果只是一个普通的响应，就将该结果放到 channel 的 哈希表中
    • 随后 channel 的 putReturns 方法会唤醒所有阻塞等待的线程，让这些线程从 哈希表中拿与自己 rid 相等的返回结果
    1. //    使用这个方法来分别处理，当前的响应是一个针对控制请求的响应，还是服务器推送的消息
    2.     private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
    3.         if(response.getType() == 0xc) {
    4. //            服务器推送来的消息数据
    5.             SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPayload());
    6. //            根据 channelId 找到对应的 channel 对象
    7.             Channel channel = channelMap.get(subScribeReturns.getChannelId());
    8.             if(channel == null) {
    9.                 throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在！channelId = " + channel.getChannelId());
    10.             }
    11. //            执行该 channel 对象内部的回调
    12. //            此处我们直接将回调方法交给线程池来执行，而不是用扫描线程来执行
    13.             callbackPool.submit(() -> {
    14.                 try {
    15.                     channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(),subScribeReturns.getBasicProperties(),
    16.                             subScribeReturns.getBody());
    17.                 } catch (MqException | IOException e) {
    18.                     e.printStackTrace();
    19.                 }
    20.             });
    21.         }else {
    22. //            当前响应是针对刚才控制请求的响应
    23.             BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());
    24. //            把这个结果放到对应的 channel 的 hash 表中
    25.             Channel channel = channelMap.get(basicReturns.getChannelId());
    26.             if(channel == null) {
    27.                 throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在！channelId = " + channel.getChannelId());
    28.             }
    29.             channel.putReturns(basicReturns);
    30.         }
    31.     }
    • 编写 发送请求 与 读取响应 的方法
    1. //    发送请求
    2.     public void writeRequest(Request request) throws IOException {
    3.         dataOutputStream.writeInt(request.getType());
    4.         dataOutputStream.writeInt(request.getLength());
    5.         dataOutputStream.write(request.getPayload());
    6.         dataOutputStream.flush();
    7.         System.out.println("[Connection] 发送请求！ type = " + request.getType() + ",length = " + request.getLength());
    8.     }
    9.  
    10. //    读取响应
    11.     public Response readResponse() throws IOException {
    12.         Response response = new Response();
    13.         response.setType(dataInputStream.readInt());
    14.         response.setLength(dataInputStream.readInt());
    15.         byte[] payload = new byte[response.getLength()];
    16.         int n = dataInputStream.read(payload);
    17.         if(n != payload.length) {
    18.             throw new IOException("读取的响应数据不完整！");
    19.         }
    20.         response.setPayload(payload);
    21.         System.out.println("[Connection] 收到响应！ type = " + response.getType() + ",length = " + response.getLength());
    22.         return response;
    23.     }
    • 编写创建 channel 的方法

    注意：

    • 我们的代码中使用了多次 UUID 
    • message 的 id，就是用 UUID 当时加了个 M- 前缀
    • 现在 channel 的 id 也是使用 UUID 此时加个 C- 前缀
    • rid 也使用 UUID 来生成，加个前缀  R-
    1. //    通过这个方法，在 Connection 中能够创建出一个 Channel
    2.     public Channel createChannel() throws IOException, InterruptedException {
    3.         String channelId = "C-" + UUID.randomUUID().toString();
    4.         Channel channel = new Channel(channelId,this);
    5. //        把这个 channel 对象放到 Connection 管理 channel 的 哈希表 中
    6.         channelMap.put(channelId,channel);
    7. //        同时也需要把 创建 channel 的这个消息也告诉服务器
    8.         boolean ok = channel.createChannel();
    9.         if(!ok) {
    10. //            服务器这里创建失败了！！整个这次创建 channel 操作不顺利！
    11. //            把刚才已经加入 hash 表的键值对，再删了
    12.             channelMap.remove(channelId);
    13.             return null;
    14.         }
    15.         return channel;
    16.     }
    • 编写释放 Connection 相关资源的方法
    1.     public void close() {
    2. //        关闭 Connection 释放上述资源
    3.         try {
    4.             callbackPool.shutdownNow();
    5.             channelMap.clear();
    6.             inputStream.close();
    7.             outputStream.close();
    8.             socket.close();
    9.         } catch (IOException e) {
    10.             e.printStackTrace();
    11.         }
    12.     }

    Channel（通道）

    • 这个类表示一个逻辑上的连接

    • 该类用于提供一系列的方法，去和服务器提供的核心 API 相对应

    • 客户端提供的这些方法，其方法内部就是发送了一个特定的请求

    具体代码编写：

    • 编写成员变量  与 构造方法
    1.     private String channelId;
    2. //    当前这个 channel 属于哪个连接的
    3.     private Connection connection;
    4. //    用来存储后续客户端收到的服务器的响应
    5.     private ConcurrentHashMap basicReturnsMap = new ConcurrentHashMap<>();
    6. //    如果当前 Channel 订阅了某个队列，就需要在此处记录下对应回调是啥，当该队列的消息返回回来的时候，调用回调
    7. //    此处约定一个 Channel 中只能有一个回调
    8.     private Consumer consumer = null;
    9.  
    10.     public Channel(String channelId,Connection connection) {
    11.         this.channelId = channelId;
    12.         this.connection = connection;
    13.     }
    • 实现 type = 0x1，即创建 channel
    • 构造请求发给服务器，随后阻塞等待，唤醒后从 basicReturnsMap 中尝试获取响应结果
    • 其余 type （0xc 除外，因为 0xc 只有响应没有请求）类型的请求与 0x1 大差不差，对着所需参数，构造即可
    1. //    在这个方法中，和服务器进行交互，告知服务器，此处客户端创建了新的 channel 了
    2.     public Boolean createChannel() throws IOException, InterruptedException {
    3. //        对于创建 Channel 来说，
    4.         BasicArguments basicArguments = new BasicArguments();
    5.         basicArguments.setChannelId(channelId);
    6.         basicArguments.setRid(generateRid());
    7.         byte[] payload = BinaryTool.toBytes(basicArguments);
    8.  
    9.         Request request = new Request();
    10.         request.setType(0x1);
    11.         request.setLength(payload.length);
    12.         request.setPayload(payload);
    13.  
    14. //        构造出完整请求之后，就可以发送这个请求了
    15.         connection.writeRequest(request);
    16. //        等待服务器的响应
    17.         BasicReturns basicReturns = waitResult(basicArguments.getRid());
    18.         return basicReturns.isOk();
    19.     }
    20.  
    21.     private String generateRid() {
    22.         return "R-" + UUID.randomUUID().toString();
    23.     }
    24.  
    25. //    期望使用这个方法来阻塞等待服务器的响应
    26.     private BasicReturns waitResult(String rid) throws InterruptedException {
    27.         BasicReturns basicReturns = null;
    28.         while ((basicReturns = basicReturnsMap.get(rid)) == null) {
    29. //            如果查询结果为 null，说明响应还没回来
    30. //            此时就需要阻塞等待
    31.             synchronized (this) {
    32.                 wait();
    33.             }
    34.         }
    35. //        读取成功之后，还需要把这个消息从 哈希表中给删除掉
    36.         basicReturnsMap.remove(rid);
    37.         System.out.println("[Channel] 获取到服务器响应！rid = " + rid);
    38.         return basicReturns;
    39.     }
    40.  
    41.     public void putReturns(BasicReturns basicReturns) {
    42.         basicReturnsMap.put(basicReturns.getRid(),basicReturns);
    43.         synchronized (this) {
    44. //            当前也不知道有多少个线程在等待上述的这个响应
    45. //            把所有的等待的线程都唤醒
    46.             notifyAll();
    47.         }
    48.     }
    • 特别注意 type = 0xa ，即 订阅消息

    • 其次值得注意的是 consumerTag 使用 channelId 来表示
    1. //    订阅消息
    2.     public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException, InterruptedException {
    3. //        先设置回调
    4.         if(this.consumer != null) {
    5.             throw new MqException("该 channel 已经设置过消费消息的回调了，不能重复设置！");
    6.         }
    7.         this.consumer = consumer;
    8.         BasicConsumeArguments basicConsumeArguments = new BasicConsumeArguments();
    9.         basicConsumeArguments.setRid(generateRid());
    10.         basicConsumeArguments.setChannelId(channelId);
    11. //        此处 ConsumerTag 也使用 channelId 来表示
    12.         basicConsumeArguments.setConsumerTag(channelId);
    13.         basicConsumeArguments.setQueueName(queueName);
    14.         basicConsumeArguments.setAutoAck(autoAck);
    15.         byte[] payload = BinaryTool.toBytes(basicConsumeArguments);
    16.  
    17.         Request request = new Request();
    18.         request.setType(0xa);
    19.         request.setLength(payload.length);
    20.         request.setPayload(payload);
    21.  
    22.         connection.writeRequest(request);
    23.         BasicReturns basicReturns = waitResult(basicConsumeArguments.getRid());
    24.         return basicReturns.isOk();
    25.     }

    针对 客户端 和 服务器 单元测试

    • 编写测试用例代码是十分重要的！
    1. package com.example.demo;
    2.  
    3. import com.example.demo.common.Consumer;
    4. import com.example.demo.common.MqException;
    5. import com.example.demo.mqclient.Channel;
    6. import com.example.demo.mqclient.Connection;
    7. import com.example.demo.mqclient.ConnectionFactory;
    8. import com.example.demo.mqserver.BrokerServer;
    9. import com.example.demo.mqserver.core.BasicProperties;
    10. import com.example.demo.mqserver.core.ExchangeType;
    11. import org.apache.tomcat.util.http.fileupload.FileUtils;
    12. import org.junit.jupiter.api.AfterEach;
    13. import org.junit.jupiter.api.Assertions;
    14. import org.junit.jupiter.api.BeforeEach;
    15. import org.junit.jupiter.api.Test;
    16. import org.springframework.boot.SpringApplication;
    17.  
    18. import java.io.File;
    19. import java.io.IOException;
    20.  
    21. public class MqClientTests {
    22.     private BrokerServer brokerServer = null;
    23.     private ConnectionFactory connectionFactory = null;
    24.     private Thread t = null;
    25.  
    26.     @BeforeEach
    27.     public void setUp() throws IOException {
    28. //        1、先启动服务器
    29.         DemoApplication.context = SpringApplication.run(DemoApplication.class);
    30.         brokerServer = new BrokerServer(9090);
    31.         t = new Thread(() -> {
    32. //        这个 start 方法会进入一个死循环，使用一个新的线程来运行 start 即可！
    33.             try {
    34.                 brokerServer.start();
    35.             } catch (IOException e) {
    36.                 e.printStackTrace();
    37.             }
    38.         });
    39.         t.start();
    40.  
    41. //        2、配置 ConnectionFactory
    42.         connectionFactory = new ConnectionFactory();
    43.         connectionFactory.setHost("127.0.0.1");
    44.         connectionFactory.setPort(9090);
    45.  
    46.     }
    47.  
    48.     @AfterEach
    49.     public void tearDown() throws IOException {
    50. //        停止服务器
    51.         brokerServer.stop();
    52. //        t.join();
    53.         DemoApplication.context.close();
    54.  
    55. //        删除必要的文件
    56.         File file = new File("./data");
    57.         FileUtils.deleteDirectory(file);
    58.  
    59.         connectionFactory = null;
    60.     }
    61.  
    62.     @Test
    63.     public void testConnection() throws IOException {
    64.         Connection connection = connectionFactory.newConnection();
    65.         Assertions.assertNotNull(connection);
    66.     }
    67.  
    68.     @Test
    69.     public void testChannel() throws IOException, InterruptedException {
    70.         Connection connection = connectionFactory.newConnection();
    71.         Assertions.assertNotNull(connection);
    72.         Channel channel = connection.createChannel();
    73.         Assertions.assertNotNull(channel);
    74.     }
    75.  
    76.     @Test
    77.     public void testExchange() throws IOException, InterruptedException {
    78.         Connection connection = connectionFactory.newConnection();
    79.         Assertions.assertNotNull(connection);
    80.         Channel channel = connection.createChannel();
    81.         Assertions.assertNotNull(channel);
    82.  
    83.         boolean ok = channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);
    84.         Assertions.assertTrue(ok);
    85.  
    86.         ok = channel.exchangeDelete("testExchange");
    87.         Assertions.assertTrue(ok);
    88.  
    89. //        此处稳妥起见，把该关闭的要进行关闭
    90.         channel.close();
    91.         connection.close();
    92.     }
    93.  
    94.     @Test
    95.     public void testQueue() throws IOException, InterruptedException{
    96.         Connection connection = connectionFactory.newConnection();
    97.         Assertions.assertNotNull(connection);
    98.         Channel channel = connection.createChannel();
    99.         Assertions.assertNotNull(channel);
    100.  
    101.         boolean ok = channel.queueDeclare("testQueue",true,false,false,null);
    102.         Assertions.assertTrue(ok);
    103.  
    104.         ok = channel.queueDelete("testQueue");
    105.         Assertions.assertTrue(ok);
    106.  
    107.         channel.close();
    108.         connection.close();
    109.     }
    110.  
    111.     @Test
    112.     public void testBinding() throws IOException, InterruptedException{
    113.         Connection connection = connectionFactory.newConnection();
    114.         Assertions.assertNotNull(connection);
    115.         Channel channel = connection.createChannel();
    116.         Assertions.assertNotNull(channel);
    117.  
    118.         boolean ok = channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);
    119.         Assertions.assertTrue(ok);
    120.  
    121.         ok = channel.queueDeclare("testQueue",true,false,false,null);
    122.         Assertions.assertTrue(ok);
    123.  
    124.         ok = channel.queueBind("testQueue","testExchange","testBindingKey");
    125.         Assertions.assertTrue(ok);
    126.  
    127.         ok = channel.queueUnbind("testQueue","testExchange");
    128.         Assertions.assertTrue(ok);
    129.  
    130.         channel.close();
    131.         connection.close();
    132.     }
    133.  
    134.     @Test
    135.     public void testMessage() throws IOException, InterruptedException, MqException {
    136.         Connection connection = connectionFactory.newConnection();
    137.         Assertions.assertNotNull(connection);
    138.         Channel channel = connection.createChannel();
    139.         Assertions.assertNotNull(channel);
    140.  
    141.         boolean ok = channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);
    142.         Assertions.assertTrue(ok);
    143.  
    144.         ok = channel.queueDeclare("testQueue",true,false,false,null);
    145.         Assertions.assertTrue(ok);
    146.  
    147.         byte[] requestBody = "hello".getBytes();
    148.         ok = channel.basicPublish("testExchange","testQueue",null,requestBody);
    149.         Assertions.assertTrue(ok);
    150.  
    151.         channel.basicConsume("testQueue", true, new Consumer() {
    152.             @Override
    153.             public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
    154.                 System.out.println("[消费数据] 开始！");
    155.                 System.out.println("consumeTag = " + consumerTag);
    156.                 System.out.println("basicProperties = " + basicProperties);
    157.                 Assertions.assertArrayEquals(requestBody,body);
    158.                 System.out.println("[消费数据] 结束！");
    159.             }
    160.         });
    161.         Assertions.assertTrue(ok);
    162.  
    163.         Thread.sleep(500);
    164.  
    165.         channel.close();
    166.         connection.close();
    167.     }
    168. }
  • 相关阅读:
    【STM32】MCU HardFault异常处理分析流程及总结（一）
    独立双样本T-Test 前 为什么要先进行列文检验(Levene‘s Test)
    关于白盒测试，这些技巧你得游刃有余~
    视觉SLAM十四讲(高翔版本)，ch1-2章节部分笔记
    nginx基础概念
    【React源码】(十七)React 算法之深度优先遍历
    智能捡乒乓球机器人
    idea创建spark项目
    Apache POI使用
    向善的力量：顺丰，如何在不确定性中寻求确定性
  • 原文地址：https://blog.csdn.net/weixin_63888301/article/details/136242153