0
0

RabbitMQ - 以 MQ 为例，手写一个 RPC 框架 demo

前言

本篇文章来自于笔者之前写过的一个系列 —— “根据源码，模拟实现 RabbitMQ” 系列，不妨可以去看看~

一、再谈自定义应用层协议

a）这个自定义应用层协议实际上就是在描述将来 客户端 和 服务器 之间通讯的消息格式长啥样

b）首先是一个 Int 类型的 type，描述了这个消息到底是用来干什么的（要调用服务器这边的哪一个服务）.

c）然后就是 payload 的数据载荷，承载着将来调用 VirtualHost 中的具体的服务所需要的参数（例如创建交换机所需要的参数就有：交换机名字、交换机类型、是否自动删除、是否持久化、扩展参数）.

因为 TCP 是面向字节流的（IO 流中主要提供的就是二进制数据的读写），因此这里不太适合使用 JSON 格式数据进行网络传输（可读性不好，效率不高），因此这里 payload 是一个 字节数组，将具体的数据序列化成 byte 数组放进来.

d）这里要注意的一点是，TCP 是面向字节流的，因此会出现粘包问题，那么为了解决这个问题，由两种办法，第一种就是约定分割符（读到指定分隔符就截止），第二种就是描述好 payload 的长度.

这里我采用的就是第二种办法，只需要在协议里面在添加一个 length 字段，用来描述 payload 的长度.

  1. import java.io.Serializable
  3. //Socket 自定义应用层协议(请求)
  4. data class Request(
  5.     val type: Int,
  6.     val length: Int,
  7.     val payload: ByteArray,
  8. ): Serializable
  10. //Socket 自定义应用层协议(响应)
  11. data class Response(
  12.     val type: Int,
  13.     val length: Int,
  14.     val payload: ByteArray,
  15. ): Serializable
  17. //基本参数(每个请求都会携带的参数，这里进行了一个封住)
  18. open class ReqBaseArguments(
  19.     open val rid: String = "",
  20.     open val channelId: String = "",
  21. ): Serializable
  23. //基本响应参数(每个响应都会携带的参数)，主要是为了应对 mq 回调响应处理
  24. open class RespBaseArguments(
  25.     open val rid: String,
  26.     open val channelId: String,
  27.     open val ok: Boolean,
  28. ): Serializable
  30. //主要的请求: 创建交换机、删除交换机、创建队列
  31. data class ExchangeDeclareReq(
  32.     val name: String,
  33.     val type: ExchangeType,
  34.     val durable: Boolean,
  35.     val autoDelete: Boolean,
  36.     val arguments: MutableMap<String, Any>,
  37.     override val rid: String,
  38.     override val channelId: String,
  39. ): ReqBaseArguments(), Serializable
  41. data class ExchangeDeleteReq(
  42.     val name: String,
  43.     override val rid: String,
  44.     override val channelId: String,
  45. ): ReqBaseArguments(), Serializable
  47. data class QueueDeclareReq(
  48.     val name: String,
  49.     val durable: Boolean,
  50.     val exclusive: Boolean,
  51.     val autoDelete: Boolean,
  52.     val arguments: MutableMap<String, Any>,
  53.     override val rid: String,
  54.     override val channelId: String,
  55. ): ReqBaseArguments(), Serializable

二、再谈 BrokerServer

a）BrokerServer 就是一个中间服务，也可以简单理解为 VirtualHost 的代理（BrokerServer 接收客户端请求，调用 VirtualHost 中具体的服务）.

b）BrokerServer 启动的时候，就会通过 accept 阻塞等待客户端这边的 TCP 连接，连接成功之后只需要为该客户端其分配一个线程，处理之后的任务.

c）此时这个线程就会处于一个死循环循环，通过 IO 流读取到 客户端 请求中的 type、length、payload ，并按照约定的格式进行解析 payload，得到具体数据（这里不仅包含了 VirtualHost 服务中所需要的具体的参数，还携带了 channelId 和 rid）

d）此时，只需要根据 IO 流中读取出的 type，调用对应 VirtualHost 中的服务即可.

e）最后再将 VirtualHost 处理后得到的响应封装成 我们约定的应用层协议格式，通过 IO 写入到流中，让客户端去读取.

  1. class BrokerServer(
  2.     port: Int
  3. ) {
  5.     private val socket = ServerSocket(port)
  6.     private val clientPool = Executors.newFixedThreadPool(5)
  8.     //key: channelId ,value: Socket
  9.     //注意：这里的 Channel 只表示一个 "逻辑" 上的连接(创建，销毁 channel)，这个 Map 是为了后台信息统计
  10.     private val channelSession = ConcurrentHashMap<String, Socket>()
  12.     private val virtualHost = VirtualHost()
  14.     fun start() {
  15.         println("[BrokerServer] 启动！")
  16.         while (true) {
  17.             val client = socket.accept()
  18.             clientPool.submit {
  19.                 clientProcess(client)
  20.             }
  21.         }
  22.     }
  24.     private fun clientProcess(client: Socket) {
  25.         println("[BrokerServer] 客户端上线！ip: ${client.inetAddress}, port: ${client.port}")
  26.         try {
  27.             client.getInputStream().use { inputStream ->
  28.                 client.getOutputStream().use { outputStream ->
  29.                     DataInputStream(inputStream).use { dataInputStream ->
  30.                         DataOutputStream(outputStream).use { dataOutputStream ->
  31.                             while (true) {
  32.                                 val request = readRequest(dataInputStream)
  33.                                 val response = process(request, client)
  34.                                 writeResponse(response, dataOutputStream)
  35.                             }
  36.                         }
  37.                     }
  38.                 }
  39.             }
  40.         } catch (e: EOFException) {
  41.             println("[BrokerServer] 客户端正常下线！ip: ${client.inetAddress}, port: ${client.port}")
  42.         } catch (e: Exception) {
  43.             println("[BrokerServer] 客户端连接异常！ip: ${client.inetAddress}, port: ${client.port}")
  44.         } finally {
  45.             client.close()
  46.             removeChannelSession(client)
  47.         }
  48.     }
  50.     private fun process(request: Request, client: Socket) = with(request) {
  51.         //1.解析请求
  52.         val req = BinaryTool.bytesToAny(payload)
  53.         //2.获取请求中的 channelId，记录和 Socket 的关系(让每个 channel 都对应自己的 Socket，类似于 Session)
  54.         val reqBase = req as ReqBaseArguments
  55.         //3.根据 type 类型执行不同的服务(创建 Channel、销毁 Channel、创建交换机、删除交换机...)
  56.         val ok = when(type) {
  57.             1 -> {
  58.                 channelSession[reqBase.channelId] = client
  59.                 println("[BrokerServer] channel 创建成功！channelId: ${reqBase.channelId}")
  60.                 true
  61.             }
  62.             2 -> {
  63.                 channelSession.remove(reqBase.channelId)
  64.                 println("[BrokerServer] channel 销毁成功！channelId: ${reqBase.channelId}")
  65.                 true
  66.             }
  67.             3 -> virtualHost.exchangeDeclare(req as ExchangeDeclareReq)
  68.             4 -> virtualHost.exchangeDelete(req as ExchangeDeleteReq)
  69.             5 -> virtualHost.queueDeclare(req as QueueDeclareReq)
  70.             //...
  71.             else -> throw RuntimeException("[BrokerServer] 客户端请求 type 非法！type: $type")
  72.         }
  73.         //4.返回响应
  74.         val respBase = RespBaseArguments(reqBase.rid, reqBase.channelId, ok)
  75.         val payload = BinaryTool.anyToBytes(respBase)
  76.         Response(type, payload.size, payload)
  77.     }
  79.     /**
  80.      * 读取客户端请求
  81.      * 使用 DataInputStream 的主要原因就是有多种读取方式，例如 readInt()、readLong()，这些都是原生 InputStream 没有的
  82.      */
  83.     private fun readRequest(dataInputStream: DataInputStream) = with(dataInputStream) {
  84.         val type = readInt()
  85.         val length = readInt()
  86.         val payload = ByteArray(length)
  87.         val n = read(payload)
  88.         if (n != length) throw RuntimeException("[BrokerServer] 读取客户端请求异常！")
  89.         Request(type, length, payload)
  90.     }
  92.     /**
  93.      * 将响应写回给客户端
  94.      */
  95.     private fun writeResponse(response: Response, outputStream: DataOutputStream) = with(outputStream) {
  96.         writeInt(response.type)
  97.         writeInt(response.length)
  98.         write(response.payload)
  99.         flush()
  100.     }
  102.     //删除所有和这个 clientSocket 有关的 Channel
  103.     private fun removeChannelSession(client: Socket) {
  104.         val channelIdList = mutableListOf<String>()
  105.         //这里不能直接删除，会破坏迭代器结构
  106.         for (entry in channelSession) {
  107.             if (entry.value == client) channelIdList.add(entry.key)
  108.         }
  109.         for (channelId in channelIdList) {
  110.             channelSession.remove(channelId)
  111.         }
  112.     }
  114. }
  1. class VirtualHost {
  3.     fun exchangeDeclare(req: ExchangeDeclareReq): Boolean {
  4.         //执行业务逻辑
  5.         //...
  6.         println("[VirtualHost] 创建交换机成功！")
  7.         return true
  8.     }
  10.     fun exchangeDelete(req: ExchangeDeleteReq): Boolean {
  11.         //执行业务逻辑
  12.         //...
  13.         println("[VirtualHost] 删除交换机成功！")
  14.         return true
  15.     }
  17.     fun queueDeclare(req: QueueDeclareReq): Boolean {
  18.         //执行业务逻辑
  19.         //...
  20.         println("[VirtualHost] 创建队列成功！")
  21.         return true
  22.     }
  24. }

三、再谈 Connection、Channel

a）一个 Connection 就是一个 TCP 连接，因此频繁 建立/断开连接（三次握手、四次挥手...）的开销也是相当大的，因此就引入了 Channel.

b）一个 Connection 下可以有多个 Channel（此处使用 map 来维护）. Channel 只是简单的表示一个逻辑上的连接，可以理解为一个大的项目下被拆分成的多个小的微服务. 实现了 TCP 连接的复用.

c）起初，我们需要先创建出 Connection 与服务端建立连接，初始化构造中只需要写一个死循环，不断的从服务端这边读取响应.

d）接着，通过 Connection 创建出 Channel 来完成具体的业务（Channel 中就提供了一系列方法，就像调用本地的方法一样，调用到远程服务器的接口）.

e）例如 Channel 中提供的创建叫交换机方法（channel.exchangeDeclare(...)），这个方法中具体要做的就是将传入的参数，封装到一个对象中，序列化成 二进制 数据，这就是将来协议中要传输的 payload. 进一步的，协议 Request 就构造出来了，通过 IO 写到流中，供服务端读取.

d）为了能够让每次请求和响应都能对的上，Channel 这里我维护了一个 map（key 是 rid、value 是具体的响应），客户端和服务端之间的每个请求和响应都会携带上这个 rid 这个参数，这样将来 Connection 客户端接受到响应的时候，就可以直接把 响应中的 rid 提取出来，交给 Channel 的 map 中（响应来之前，Channel 一直阻塞等待，直到响应来了 -> 能通过 rid 从 map 中得到）.

  1. class ConnectionFactory(
  2.     private val host: String,
  3.     private val port: Int,
  4. ) {
  6.     fun newConnection() = Connection(host, port)
  8. }
  1. class Connection(
  2.     ip: String,
  3.     port: Int,
  4. ) {
  6.     private val socket = Socket(ip, port)
  7.     private val channelMap = ConcurrentHashMap<String, Channel>()
  8.     //下述这样提前创建好，是为了将来 Channel 在读写请求的时候的方便(Channel 就不用获取输入输出流了)
  9.     private val inputStream = socket.getInputStream()
  10.     private val outputStream = socket.getOutputStream()
  11.     private val dataInputStream = DataInputStream(inputStream)
  12.     private val dataOutputStream = DataOutputStream(outputStream)
  14.     init {
  15.         //此线程负责不停的从服务器这边获取响应
  16.          Thread {
  17.              try {
  18.                  while (!socket.isClosed) {
  19.                      //读取服务器响应
  20.                      val resp = readResp()
  21.                      //将响应交给对应的 Channel
  22.                      putRespToChannel(resp)
  23.                  }
  24.              } catch (e: SocketException) {
  25.                  println("[Connection] 客户端正常断开连接")
  26.              } catch (e: Exception) {
  27.                  println("[Connection] 客户端异常断开连接")
  28.                  e.printStackTrace()
  29.              }
  30.          }.start()
  31.     }
  33.     /**
  34.      * 将客户端 Connection 接收到的请求，交给对应的 Channel 处理(此时 Channel 还在阻塞等待服务端响应)
  35.      */
  36.     private fun putRespToChannel(resp: Response) {
  37.         //这里由于不涉及回调，所以每个 type 类型的响应都长一样，就按照一样的方式解析了
  38.         val baseResp = BinaryTool.bytesToAny(resp.payload) as RespBaseArguments
  39.         val channel = channelMap[baseResp.channelId]
  40.             ?: throw RuntimeException("[Connection] 该响应对应的 Channel 不存在！channelId: ${baseResp.channelId}")
  41.         //将响应交给 Channel
  42.         channel.notifyResp(baseResp)
  43.     }
  45.     /**
  46.      * 创建 Channel
  47.      */
  48.     fun createChannel(): Channel { //1.创建 Channel，保存到 map 种
  49.         val channelId = "C-${UUID.randomUUID()}"
  50.         val channel = Channel(channelId, this)
  51.         channelMap[channelId] = channel
  52.         //2.告知服务端 Channel 创建
  53.         val ok = channel.createChannel()
  54.         //3.如果 Channel 创建不成功，客户端这边也应该要删除对应的 Channel 信息
  55.         if (!ok) channelMap.remove(channelId)
  56.         return channel
  57.     }
  59.     private fun readResp() = with(dataInputStream) {
  60.         val type = readInt()
  61.         val length = readInt()
  62.         val payload = ByteArray(length)
  63.         val n = read(payload)
  64.         if (n != length) throw RuntimeException("[Connection] 客户端读取响应异常！")
  65.         Response(type, length, payload)
  66.     }
  68.     fun writeReq(request: Request) = with(dataOutputStream) {
  69.         writeInt(request.type)
  70.         writeInt(request.length)
  71.         write(request.payload)
  72.         flush()
  73.     }
  75. }
  1. class Channel(
  2.     private val channelId: String,
  3.     private val connection: Connection,  //自己当前属于哪个 Channel
  4. ) {
  6.     //key: rid(为了能让每个 Channel 对应上自己的响应)
  7.     //value: RespBaseArguments(具体的响应)
  8.     //当 Connection 的扫描线程接收到响应之后，就会将响应传给这个 map
  9.     private val ridRespMap = ConcurrentHashMap<String, RespBaseArguments>()
  10.     //这个锁是用来阻塞等待服务端响应的(避免轮询)，当服务端传来响应时，Connection 就会唤醒锁
  11.     private val locker = Object()
  13.     private fun generateRid() = "R-${UUID.randomUUID()}"
  15.     private fun waitResp(rid: String): RespBaseArguments {
  16.         val respBase: RespBaseArguments
  17.         while (ridRespMap[rid] == null) { // 如果为空，说明此时服务端还没有传来响应
  18.             synchronized(locker) { //为了避免轮询，就让其阻塞等待
  19.                 locker.wait()
  20.             }
  21.         }
  22.         //出了这个循环，那么 ridRespMap[rid] 一定不为空
  23.         return ridRespMap[rid]!!
  24.     }
  26.     fun notifyResp(respBase: RespBaseArguments) {
  27.         ridRespMap[respBase.rid] = respBase
  28.         synchronized(locker) {
  29.             //当前也不直到有多少线程在等待响应，就全部唤醒
  30.             locker.notifyAll()
  31.         }
  32.     }
  34.     /**
  35.      * 创建 Channel
  36.      */
  37.     fun createChannel(): Boolean {
  38.         //1.创建基本请求
  39.         val reqBase = ReqBaseArguments(
  40.             rid = generateRid(),
  41.             channelId = channelId
  42.         )
  43.         //2.构造 TCP 通信请求
  44.         val payload = BinaryTool.anyToBytes(reqBase)
  45.         val req = Request(
  46.             type = 1,
  47.             length = payload.size,
  48.             payload = payload
  49.         )
  50.         //3.发送请求
  51.         connection.writeReq(req)
  52.         //4.等待客户端响应
  53.         val respBase = waitResp(reqBase.rid)
  54.         return respBase.ok
  55.     }
  57.     fun removeChannel(): Boolean {
  58.         //1.创建基本请求
  59.         val reqBase = ReqBaseArguments(
  60.             rid = generateRid(),
  61.             channelId = channelId
  62.         )
  63.         //2.构造 TCP 通信请求
  64.         val payload = BinaryTool.anyToBytes(reqBase)
  65.         val req = Request(
  66.             type = 2,
  67.             length = payload.size,
  68.             payload = payload
  69.         )
  70.         //3.发送请求
  71.         connection.writeReq(req)
  72.         //4.等待客户端响应
  73.         val respBase = waitResp(reqBase.rid)
  74.         return respBase.ok
  75.     }
  77.     fun exchangeDeclare(
  78.         name: String,
  79.         type: ExchangeType,
  80.         durable: Boolean,
  81.         autoDelete: Boolean,
  82.         arguments: MutableMap<String, Any>,
  83.     ): Boolean {
  84.         val exchangeDeclareReq = ExchangeDeclareReq(
  85.             name = name,
  86.             type = type,
  87.             durable = durable,
  88.             autoDelete = autoDelete,
  89.             arguments = arguments,
  90.             rid = generateRid(),
  91.             channelId = channelId,
  92.         )
  93.         val payload = BinaryTool.anyToBytes(exchangeDeclareReq)
  94.         val req = Request(
  95.             type = 3,
  96.             length = payload.size,
  97.             payload = payload,
  98.         )
  99.         connection.writeReq(req)
  100.         val respBase = waitResp(exchangeDeclareReq.rid)
  101.         return respBase.ok
  102.     }
  104.     fun exchangeDelete(name: String): Boolean {
  105.         val exchangeDeleteReq = ExchangeDeleteReq(
  106.             name = name,
  107.             rid = generateRid(),
  108.             channelId = channelId,
  109.         )
  110.         val payload = BinaryTool.anyToBytes(exchangeDeleteReq)
  111.         val req = Request(
  112.             type = 4,
  113.             length = payload.size,
  114.             payload = payload,
  115.         )
  116.         connection.writeReq(req)
  117.         val respBase = waitResp(exchangeDeleteReq.rid)
  118.         return respBase.ok
  119.     }
  121.     fun queueDeclare(
  122.         name: String,
  123.         durable: Boolean,
  124.         exclusive: Boolean,
  125.         autoDelete: Boolean,
  126.         arguments: MutableMap<String, Any>,
  127.     ): Boolean {
  128.         val queueDeclareReq = QueueDeclareReq(
  129.             name = name,
  130.             durable = durable,
  131.             exclusive = exclusive,
  132.             autoDelete = autoDelete,
  133.             arguments = arguments,
  134.             rid = generateRid(),
  135.             channelId = channelId,
  136.         )
  137.         val payload = BinaryTool.anyToBytes(queueDeclareReq)
  138.         val req = Request(
  139.             type = 5,
  140.             length = payload.size,
  141.             payload = payload,
  142.         )
  143.         connection.writeReq(req)
  144.         val resp = waitResp(queueDeclareReq.rid)
  145.         return resp.ok
  146.     }
  148. }

四、Demo

a）启动服务器

  1. fun main() {
  2.     val server = BrokerServer(9000)
  3.     server.start()
  4. }

b）客户端连接

  1. class Test2 {
  2. }
  4. fun main() {
  5.     val factory = ConnectionFactory("127.0.0.1", 9000)
  6.     val connection = factory.newConnection()
  7.     val channel = connection.createChannel()
  9.     val ok1 = channel.createChannel()
  10.     val ok2 = channel.exchangeDeclare("e1", ExchangeType.DIRECT, false, false, mutableMapOf())
  11.     val ok3 = channel.removeChannel()
  13.     println("ok1: $ok1, ok2: $ok2, ok3: $ok3")
  14. }

标签： rabbitmq rpc 网络
本文转载自: https://blog.csdn.net/CYK_byte/article/details/138559976
版权归原作者 陈亦康 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ - 以 MQ 为例，手写一个 RPC 框架 demo”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
前言