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

前言

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

二、再谈 BrokerServer

三、再谈 Connection、Channel

四、Demo

a）启动服务器

b）客户端连接

前言

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

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

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

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

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

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

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

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

import java.io.Serializable//Socket 自定义应用层协议(请求)
data class Request(val type: Int,val length: Int,val payload: ByteArray,
): Serializable//Socket 自定义应用层协议(响应)
data class Response(val type: Int,val length: Int,val payload: ByteArray,
): Serializable//基本参数(每个请求都会携带的参数，这里进行了一个封住)
open class ReqBaseArguments(open val rid: String = "",open val channelId: String = "",
): Serializable//基本响应参数(每个响应都会携带的参数)，主要是为了应对 mq 回调响应处理
open class RespBaseArguments(open val rid: String,open val channelId: String,open val ok: Boolean,
): Serializable//主要的请求: 创建交换机、删除交换机、创建队列
data class ExchangeDeclareReq(val name: String,val type: ExchangeType,val durable: Boolean,val autoDelete: Boolean,val arguments: MutableMap<String, Any>,override val rid: String,override val channelId: String,
): ReqBaseArguments(), Serializabledata class ExchangeDeleteReq(val name: String,override val rid: String,override val channelId: String,
): ReqBaseArguments(), Serializabledata class QueueDeclareReq(val name: String,val durable: Boolean,val exclusive: Boolean,val autoDelete: Boolean,val arguments: MutableMap<String, Any>,override val rid: String,override val channelId: String,
): 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 写入到流中，让客户端去读取.

class BrokerServer(port: Int
) {private val socket = ServerSocket(port)private val clientPool = Executors.newFixedThreadPool(5)//key: channelId ,value: Socket//注意：这里的 Channel 只表示一个 "逻辑" 上的连接(创建，销毁 channel)，这个 Map 是为了后台信息统计private val channelSession = ConcurrentHashMap<String, Socket>()private val virtualHost = VirtualHost()fun start() {println("[BrokerServer] 启动！")while (true) {val client = socket.accept()clientPool.submit {clientProcess(client)}}}private fun clientProcess(client: Socket) {println("[BrokerServer] 客户端上线！ip: ${client.inetAddress}, port: ${client.port}")try {client.getInputStream().use { inputStream ->client.getOutputStream().use { outputStream ->DataInputStream(inputStream).use { dataInputStream ->DataOutputStream(outputStream).use { dataOutputStream ->while (true) {val request = readRequest(dataInputStream)val response = process(request, client)writeResponse(response, dataOutputStream)}}}}}} catch (e: EOFException) {println("[BrokerServer] 客户端正常下线！ip: ${client.inetAddress}, port: ${client.port}")} catch (e: Exception) {println("[BrokerServer] 客户端连接异常！ip: ${client.inetAddress}, port: ${client.port}")} finally {client.close()removeChannelSession(client)}}private fun process(request: Request, client: Socket) = with(request) {//1.解析请求val req = BinaryTool.bytesToAny(payload)//2.获取请求中的 channelId，记录和 Socket 的关系(让每个 channel 都对应自己的 Socket，类似于 Session)val reqBase = req as ReqBaseArguments//3.根据 type 类型执行不同的服务(创建 Channel、销毁 Channel、创建交换机、删除交换机...)val ok = when(type) {1 -> {channelSession[reqBase.channelId] = clientprintln("[BrokerServer] channel 创建成功！channelId: ${reqBase.channelId}")true}2 -> {channelSession.remove(reqBase.channelId)println("[BrokerServer] channel 销毁成功！channelId: ${reqBase.channelId}")true}3 -> virtualHost.exchangeDeclare(req as ExchangeDeclareReq)4 -> virtualHost.exchangeDelete(req as ExchangeDeleteReq)5 -> virtualHost.queueDeclare(req as QueueDeclareReq)//...else -> throw RuntimeException("[BrokerServer] 客户端请求 type 非法！type: $type")}//4.返回响应val respBase = RespBaseArguments(reqBase.rid, reqBase.channelId, ok)val payload = BinaryTool.anyToBytes(respBase)Response(type, payload.size, payload)}/*** 读取客户端请求* 使用 DataInputStream 的主要原因就是有多种读取方式，例如 readInt()、readLong()，这些都是原生 InputStream 没有的*/private fun readRequest(dataInputStream: DataInputStream) = with(dataInputStream) {val type = readInt()val length = readInt()val payload = ByteArray(length)val n = read(payload)if (n != length) throw RuntimeException("[BrokerServer] 读取客户端请求异常！")Request(type, length, payload)}/*** 将响应写回给客户端*/private fun writeResponse(response: Response, outputStream: DataOutputStream) = with(outputStream) {writeInt(response.type)writeInt(response.length)write(response.payload)flush()}//删除所有和这个 clientSocket 有关的 Channelprivate fun removeChannelSession(client: Socket) {val channelIdList = mutableListOf<String>()//这里不能直接删除，会破坏迭代器结构for (entry in channelSession) {if (entry.value == client) channelIdList.add(entry.key)}for (channelId in channelIdList) {channelSession.remove(channelId)}}}

class VirtualHost {fun exchangeDeclare(req: ExchangeDeclareReq): Boolean {//执行业务逻辑//...println("[VirtualHost] 创建交换机成功！")return true}fun exchangeDelete(req: ExchangeDeleteReq): Boolean {//执行业务逻辑//...println("[VirtualHost] 删除交换机成功！")return true}fun queueDeclare(req: QueueDeclareReq): Boolean {//执行业务逻辑//...println("[VirtualHost] 创建队列成功！")return true}}

三、再谈 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 中得到）.

class ConnectionFactory(private val host: String,private val port: Int,
) {fun newConnection() = Connection(host, port)}

class Connection(ip: String,port: Int,
) {private val socket = Socket(ip, port)private val channelMap = ConcurrentHashMap<String, Channel>()//下述这样提前创建好，是为了将来 Channel 在读写请求的时候的方便(Channel 就不用获取输入输出流了)private val inputStream = socket.getInputStream()private val outputStream = socket.getOutputStream()private val dataInputStream = DataInputStream(inputStream)private val dataOutputStream = DataOutputStream(outputStream)init {//此线程负责不停的从服务器这边获取响应Thread {try {while (!socket.isClosed) {//读取服务器响应val resp = readResp()//将响应交给对应的 ChannelputRespToChannel(resp)}} catch (e: SocketException) {println("[Connection] 客户端正常断开连接")} catch (e: Exception) {println("[Connection] 客户端异常断开连接")e.printStackTrace()}}.start()}/*** 将客户端 Connection 接收到的请求，交给对应的 Channel 处理(此时 Channel 还在阻塞等待服务端响应)*/private fun putRespToChannel(resp: Response) {//这里由于不涉及回调，所以每个 type 类型的响应都长一样，就按照一样的方式解析了val baseResp = BinaryTool.bytesToAny(resp.payload) as RespBaseArgumentsval channel = channelMap[baseResp.channelId]?: throw RuntimeException("[Connection] 该响应对应的 Channel 不存在！channelId: ${baseResp.channelId}")//将响应交给 Channelchannel.notifyResp(baseResp)}/*** 创建 Channel*/fun createChannel(): Channel { //1.创建 Channel，保存到 map 种val channelId = "C-${UUID.randomUUID()}"val channel = Channel(channelId, this)channelMap[channelId] = channel//2.告知服务端 Channel 创建val ok = channel.createChannel()//3.如果 Channel 创建不成功，客户端这边也应该要删除对应的 Channel 信息if (!ok) channelMap.remove(channelId)return channel}private fun readResp() = with(dataInputStream) {val type = readInt()val length = readInt()val payload = ByteArray(length)val n = read(payload)if (n != length) throw RuntimeException("[Connection] 客户端读取响应异常！")Response(type, length, payload)}fun writeReq(request: Request) = with(dataOutputStream) {writeInt(request.type)writeInt(request.length)write(request.payload)flush()}}

class Channel(private val channelId: String,private val connection: Connection,  //自己当前属于哪个 Channel
) {//key: rid(为了能让每个 Channel 对应上自己的响应)//value: RespBaseArguments(具体的响应)//当 Connection 的扫描线程接收到响应之后，就会将响应传给这个 mapprivate val ridRespMap = ConcurrentHashMap<String, RespBaseArguments>()//这个锁是用来阻塞等待服务端响应的(避免轮询)，当服务端传来响应时，Connection 就会唤醒锁private val locker = Object()private fun generateRid() = "R-${UUID.randomUUID()}"private fun waitResp(rid: String): RespBaseArguments {val respBase: RespBaseArgumentswhile (ridRespMap[rid] == null) { // 如果为空，说明此时服务端还没有传来响应synchronized(locker) { //为了避免轮询，就让其阻塞等待locker.wait()}}//出了这个循环，那么 ridRespMap[rid] 一定不为空return ridRespMap[rid]!!}fun notifyResp(respBase: RespBaseArguments) {ridRespMap[respBase.rid] = respBasesynchronized(locker) {//当前也不直到有多少线程在等待响应，就全部唤醒locker.notifyAll()}}/*** 创建 Channel*/fun createChannel(): Boolean {//1.创建基本请求val reqBase = ReqBaseArguments(rid = generateRid(),channelId = channelId)//2.构造 TCP 通信请求val payload = BinaryTool.anyToBytes(reqBase)val req = Request(type = 1,length = payload.size,payload = payload)//3.发送请求connection.writeReq(req)//4.等待客户端响应val respBase = waitResp(reqBase.rid)return respBase.ok}fun removeChannel(): Boolean {//1.创建基本请求val reqBase = ReqBaseArguments(rid = generateRid(),channelId = channelId)//2.构造 TCP 通信请求val payload = BinaryTool.anyToBytes(reqBase)val req = Request(type = 2,length = payload.size,payload = payload)//3.发送请求connection.writeReq(req)//4.等待客户端响应val respBase = waitResp(reqBase.rid)return respBase.ok}fun exchangeDeclare(name: String,type: ExchangeType,durable: Boolean,autoDelete: Boolean,arguments: MutableMap<String, Any>,): Boolean {val exchangeDeclareReq = ExchangeDeclareReq(name = name,type = type,durable = durable,autoDelete = autoDelete,arguments = arguments,rid = generateRid(),channelId = channelId,)val payload = BinaryTool.anyToBytes(exchangeDeclareReq)val req = Request(type = 3,length = payload.size,payload = payload,)connection.writeReq(req)val respBase = waitResp(exchangeDeclareReq.rid)return respBase.ok}fun exchangeDelete(name: String): Boolean {val exchangeDeleteReq = ExchangeDeleteReq(name = name,rid = generateRid(),channelId = channelId,)val payload = BinaryTool.anyToBytes(exchangeDeleteReq)val req = Request(type = 4,length = payload.size,payload = payload,)connection.writeReq(req)val respBase = waitResp(exchangeDeleteReq.rid)return respBase.ok}fun queueDeclare(name: String,durable: Boolean,exclusive: Boolean,autoDelete: Boolean,arguments: MutableMap<String, Any>,): Boolean {val queueDeclareReq = QueueDeclareReq(name = name,durable = durable,exclusive = exclusive,autoDelete = autoDelete,arguments = arguments,rid = generateRid(),channelId = channelId,)val payload = BinaryTool.anyToBytes(queueDeclareReq)val req = Request(type = 5,length = payload.size,payload = payload,)connection.writeReq(req)val resp = waitResp(queueDeclareReq.rid)return resp.ok}}

四、Demo

a）启动服务器

fun main() {val server = BrokerServer(9000)server.start()
}

b）客户端连接

class Test2 {
}fun main() {val factory = ConnectionFactory("127.0.0.1", 9000)val connection = factory.newConnection()val channel = connection.createChannel()val ok1 = channel.createChannel()val ok2 = channel.exchangeDeclare("e1", ExchangeType.DIRECT, false, false, mutableMapOf())val ok3 = channel.removeChannel()println("ok1: $ok1, ok2: $ok2, ok3: $ok3")
}