【noob-rpc】⑧扩展版RPC-自定义协议

holic-x...大约 19 分钟

【noob-rpc】⑧扩展版RPC-自定义协议

扩展核心

【1】自定义协议格式梳理（协议消息、相关协议数据字段枚举）

【2】基于Vert.x的TCP实现（参考基于Vert.x的HTTP实现思路进行构建）：先从demo（server、client）理解TCP协议的请求响应，然后在RPC框架中引入（服务提供者：server引用自定义的TcpServerHandler；服务消费者在ServerProxy中按照TCP协议规则处理响应）

【3】解决半包、粘包问题：使用Vert.x的RecordParse

【4】装饰者模式的场景应用：对半包、粘包方法进行封装（基于Handler进行装饰对buffer进行处理，引入TcpBufferHandlerWrapper）、修改ServiceProxy中TCP响应处理（将响应处理方法放在VertxTcpClient实现）

需求分析

RPC协议梳理

目前构建的RPC框架使用Vert x的HttpServer作为服务提供者的服务器，代码实现比较简单，其底层网络传输使用的是HTTP协议。

有时候可能会把HTTP和RPC理解为同一类技术，但HTTP只是RPC框架网络传输的一种可选方式罢了。

此处思考一个问题：使用HTTP协议会有什么问题么?或者说，有没有更好的选择?

一般情况下，RPC框架会比较注重性能，而HTTP协议中的头部信息、请求响应格式较“重”，会影响网络传输性能。

举个例子，利用浏览器网络控制台随便查看一个请求，能看到大量的请求和响应标头。

通过自定义一套PRC协议：利用TCP等传输层协议，自定义请求响应结构，实现性能更高、更为灵活、更安全的RPC框架。

RPC协议设计方案

自定义RPC协议可以分为2大核心部分：自定义网络传输、自定义消息结构

【1】网络传输设计

网络传输设计的目标是:选择-个能够高性能通信的网络协议和传输方式。

HTTP协议的头信息是比较大的，会影响传输性能。但其实除了这点外, HTTP 本身属于无状态协议，这意味着每个HTTP请求都是独立的，每次请求响应都要重新建立和关闭连接,也会影响性能。考虑到这点，在HTTP/1.1引入了持久连接(Keep-Alive) ，允许在单个TCP连接上发送多个HTTP请求和响应，避兔了每次请求都要重新建立和关闭连接的开销。虽然如此，HTTP 本身是应用层协议,目前现在设计的RPC协议也是应用层协议,性能肯定是不如底层(传输层)的TCP协议要高的。所以我们想要追求更高的性能，还是选择使用TCP协议完成网络传输，有更多的自主设计空间。

【2】消息结构设计

消息结构设计的目标是：用最少的空间传递要的信息。

（1）如何使用最少的空间呢?

之前接触到的数据类型可能都是整型、长整型、浮点数类型等等,这些类型其实都比较“重”，占用的字节数较多。比如整型要用4个字节、32 个bit位

在自定义消息结构时，想要节省空间，就要尽可能使用更轻量的类型，比如byte字节类型，只占用1字节、8个bit位。但需要注意的是，Java 中实现bit位运算拼接相对比较麻烦，所以权衡开发成本，在设计消息结构时，尽量给每个数据凑到整个字节。

（2）消息内需要哪些信息呢?

可以从之前的HTTP请求方式中，分析HTTP请求结构，得到RPC消息所需的信息：

魔数：作用是安全校验，防止服务器处理了非框架发来的乱七八糟的消息(类似HTTPS的安全证书)
版本号：保证请求和响应的一致性(类似HTTP协议有1.0/2.0 等版本)
列化方式：来告诉服务端和客户端如何解析数据(类似HTTP的Content- Type内容类型)
类型：标识是请求还是响应?或者是心跳检测等其他用途。(类似 HTTP有请求头和响应头)
状态：如果是响应,记录响应的结果(类似HTTP的200状态代码)

此外，还需要有请求id，唯一标识某个请求，因为TCP是双向通信的，需要有个唯一标识来追踪每个请求。

请求体：要发送body内容数据，类似于之前HTTP请求中发送的RpcRequest

如果是HTTP这种协议，有专门的key/value结构，很容易找到完整的body数据。但基于TCP协议，想要获取到完整的body内容数据，就需要一些“小心思”了，因为TCP协议本身会存在半包和粘包问题，每次传输的数据可能是不完整的、所以需要在消息头中新增一个字段请求体数据长度，保证能够完整地获取body内容信息。

基于以上的思考，可以得到最终的消息结构设计，如下图：

实际上，这些数据应该是紧凑的，请求头信息总长17个字节。也就是说，上述消息结构, 本质上就是拼接在一起的一个字节数组。后续实现时，需要有消息编码器和消息解码器，编码器先new一个空的Buffer缓冲区，然后按照顺序向缓冲区依次写入这些数据；解码器在读取时也按照顺序依次读取，就能还原出编码前的数据。

通过这种约定的方式，我们就不用记录头信息了。比如magic魔数，不用存储“magic"这个字符串，而是读取第一个字节(前8bit) 就能获取到。

如果学过Redis底层，会发现很多数据结构都是这种设计。（如果是第一次设计协议，或者经验不足，可以先去学一下优秀开源框架的协议设计，这样不会说毫无头绪）例如参考Dubbo的协议设计：

实现步骤

构建步骤说明

【1】新建protocol包，存放所有和自定义协议相关的代码（协议消息类ProtocolMessage、协议常量类ProtocolConstant、消息字段枚举类ProtocolMessageStatusEnum、消息类型枚举ProtocolMessageTypeEnum、序列化器枚举ProtocolMessageSerializerEnum）

【2】网络传输相关：server.tcp包

1.protocol包

类	说明
ProtocolMessage	协议消息类
ProtocolConstant	协议常量类
ProtocolMessageStatusEnum	协议消息字段枚举类
ProtocolMessageTypeEnum	协议消息类型枚举
ProtocolMessageSerializerEnum	序列化器枚举

ProtocolMessage

ProtocolMessage：将消息头单独封装为一个内部类，消息体可以使用泛型类型

ProtocolConstant

ProtocolConstant：记录了和自定义协议有关的关键信息，例如消息头、魔数、版本号信息等

ProtocolMessageStatusEnum

ProtocolMessageStatusEnum：消息字段枚举类相关：协议状态枚举（暂定成功、请求失败、响应失败）

ProtocolMessageTypeEnum

ProtocolMessageTypeEnum：协议消息类型枚举（请求、响应、心跳、其他等）

ProtocolMessageSerializerEnum

ProtocolMessageSerializerEnum：序列化器枚举（和RPC框架支持的序列化器对应）

2.server.tcp包

目前RPC框架使用了高性能的Vert.x作为网络传输服务器，之前用的是HttpServer。同样，Vert.x 也支持TCP服务器，相比于Netty或者自写Socket代码，更加简单易用。

类	说明
VertxTcpServer	TCP服务器实现：类似于VertxHttpServer实现
VertxTcpClient	TCP客户端实现：请求服务端

测试说明：先后启动服务端、客户端，随后查看控制台输出信息

3.编码器和解码器

基于上面的实现中，Vert.x的TCP服务器收发的消息是Buffer 类型，不能直接写入一一个对象。因此，需要自定义一个编码器和解码器，将Java的消息对象和Buffer进行相互转换。

http请求响应处理：从body处理器中获取到body字节数组，再通过序列化（反序列化）得到RpcRequest/RpcResponse对象。使用TCP服务器后调整为从Buffer中获取字节数组，然后再转码为RpcRequest/RpcResponse对象（相关处理流程都是可以复用的）

实现：protocol/ProtocolMessageEncoder、protocol/ProtocolMessageDecoder

类	说明
ProtocolMessageEncoder	消息编码器
ProtocolMessageDecoder	消息解码器

ProtocolMessageEncoder

/**
 * 协议消息编码器
 */
public class ProtocolMessageEncoder {

    /**
     * 编码
     * @param protocolMessage
     * @return
     * @throws IOException
     */
    public static Buffer encode(ProtocolMessage<?> protocolMessage) throws IOException {
        if (protocolMessage == null || protocolMessage.getHeader() == null) {
            return Buffer.buffer();
        }
        ProtocolMessage.Header header = protocolMessage.getHeader();
        // 依次向缓冲区写入字节
        Buffer buffer = Buffer.buffer();
        buffer.appendByte(header.getMagic());
        buffer.appendByte(header.getVersion());
        buffer.appendByte(header.getSerializer());
        buffer.appendByte(header.getType());
        buffer.appendByte(header.getStatus());
        buffer.appendLong(header.getRequestId());
        // 获取序列化器
        ProtocolMessageSerializerEnum serializerEnum = ProtocolMessageSerializerEnum.getEnumByKey(header.getSerializer());
        if (serializerEnum == null) {
            throw new RuntimeException("序列化协议不存在");
        }
        Serializer serializer = SerializerFactory.getInstance(serializerEnum.getValue());
        byte[] bodyBytes = serializer.serialize(protocolMessage.getBody());
        // 写入 body 长度和数据
        buffer.appendInt(bodyBytes.length);
        buffer.appendBytes(bodyBytes);
        return buffer;
    }
}

ProtocolMessageDecoder

/**
 * 协议消息解码器
 */
public class ProtocolMessageDecoder {

    /**
     * 解码
     * @param buffer
     * @return
     * @throws IOException
     */
    public static ProtocolMessage<?> decode(Buffer buffer) throws IOException {
        // 分别从指定位置读出 Buffer
        ProtocolMessage.Header header = new ProtocolMessage.Header();
        byte magic = buffer.getByte(0);
        // 校验魔数
        if (magic != ProtocolConstant.PROTOCOL_MAGIC) {
            throw new RuntimeException("消息 magic 非法");
        }
        header.setMagic(magic);
        header.setVersion(buffer.getByte(1));
        header.setSerializer(buffer.getByte(2));
        header.setType(buffer.getByte(3));
        header.setStatus(buffer.getByte(4));
        header.setRequestId(buffer.getLong(5));
        header.setBodyLength(buffer.getInt(13));
        // 解决粘包问题，只读指定长度的数据
        byte[] bodyBytes = buffer.getBytes(17, 17 + header.getBodyLength());
        // 解析消息体
        ProtocolMessageSerializerEnum serializerEnum = ProtocolMessageSerializerEnum.getEnumByKey(header.getSerializer());
        if (serializerEnum == null) {
            throw new RuntimeException("序列化消息的协议不存在");
        }
        Serializer serializer = SerializerFactory.getInstance(serializerEnum.getValue());
        ProtocolMessageTypeEnum messageTypeEnum = ProtocolMessageTypeEnum.getEnumByKey(header.getType());
        if (messageTypeEnum == null) {
            throw new RuntimeException("序列化消息的类型不存在");
        }
        switch (messageTypeEnum) {
            case REQUEST:
                RpcRequest request = serializer.deserialize(bodyBytes, RpcRequest.class);
                return new ProtocolMessage<>(header, request);
            case RESPONSE:
                RpcResponse response = serializer.deserialize(bodyBytes, RpcResponse.class);
                return new ProtocolMessage<>(header, response);
            case HEART_BEAT:
            case OTHERS:
            default:
                throw new RuntimeException("暂不支持该消息类型");
        }
    }

}

测试

4.请求处理器（服务提供者）

可以使用ntty的pipeline组合多个handler (比如编码=>解码=>请求1响应处理)

请求处理器的作用是接受请求，然后通过反射调用服务实现类。类似之前的HttpServerHandler，开发一个TcpServerHandler用于处理请求。

和HttpServerHandler的区别：只是在获取请求、写入响应的方式上，需要调用上述编码器和解码器。通过实现Vert.x提供的Handler<NetSocket> 接口，可以定义TCP请求处理器

TcpServerHandler

public class TcpServerHandler implements Handler<NetSocket> {
    /**
     * 处理请求
     *
     * @param netSocket the event to handle
     */
    @Override
    public void handle(NetSocket netSocket) {
        //处理连接
        netSocket.handler(buffer -> {
            //接受请求，解码
            ProtocolMessage<RpcRequest> protocolMessage;
            try {
                protocolMessage = (ProtocolMessage<RpcRequest>) ProtocolMessageDecoder.decode(buffer);
            } catch (IOException e) {
                throw new RuntimeException(" 协议消息解码错误");
            }
            RpcRequest rpcRequest = protocolMessage.getBody();

            //处理请求
            //构造响应结果对象
            RpcResponse rpcResponse = new RpcResponse();
            try {
                //获取要调用的服务实现类，通过反射调用
                Class<?> implClass = LocalRegistry.get(rpcRequest.getServiceName());
                Method method = implClass.getMethod(rpcRequest.getMethodName(), rpcRequest.getParameterTypes());
                Object result = method.invoke(implClass.newInstance(), rpcRequest.getArgs());
                //封装返回结果
                rpcResponse.setData(result);
                rpcResponse.setDataType(method.getReturnType(); rpcResponse.setMessage(" ok");
            } catch (Exception e) {
                e.printStackTrace();
                rpcResponse.setMessage(e.getMessage());
                rpcResponse.setException(e);
            }

            // 发送响应，编码
            ProtocolMessage.Header header = protocolMessage.getHeader();
            header.setType((byte) ProtocolMessageTypeEnum.RESPONSE.getKey();
            ProtocolMessage<RpcResponse> responseProtocolMessage = new ProtocolMessage<>(header, rpcResponse);
            try {
                Buffer encode = ProtocolMessageEncoder.encode(responseProtocolMessage);
                netSocket.write(encode);
            } catch (IOException e) {
                throw new RuntimeException(" 协议消息编码错误");
            }
        });
    }
}

5.请求发送（服务消费者）

ServiceProxy

调整服务代理ServiceProxy消费者发送请求的代码，将http调整tcp请求

核心TCP处理思路（关注核心，后处理业务逻辑）

public class ServiceProxy implements InvocationHandler {
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    
      		 CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
            netClient.connect(xxxx, result -> {
                // ....... 业务逻辑处理完成响应（解码请求信息进行处理、编码响应数据）.......
              	responseFuture.complete(rpcResponseProtocolMessage.getBody());
            });
					 // 阻塞，直到响应完成才会继续向下执行
            RpcResponse rpcResponse = responseFuture.get();
            // 关闭TCP连接
            netClient.close();
    }
}

测试

修改CoreProviderSample服务提供者，将原有http服务启动调整为tcp服务启动（此处需注意需要修改VertxTcpServer的处理器为自定义的TcpServerHandler，统一请求和响应的编码解码，否则如果消费者那边ServiceProxy做了处理而VertxTcpServer还是默认之前写的测试方法，就会出错（例如验证魔法值不合法等等））

测试：CoreProviderSample服务提供者、EasyConsumerSample服务消费者

此处的CoreProviderSample需要将原有http启动方式调整为TCP启动，随后先后启动服务提供者、服务消费者再次连接尝试

// 启动web服务（从RPC框架中的全局配置中获取端口）
// HttpServer httpServer = new VertxHttpServer();
// httpServer.doStart(RpcApplication.getRpcConfig().getServerPort());

// TCP协议方式启动
VertxTcpServer vertxTcpServer = new VertxTcpServer();
vertxTcpServer.doStart(RpcApplication.getRpcConfig().getServerPort());

启动测试，主要关注服务消费者可否正常获取到对应的参数信息即可。如果出现异常则一步步进行排查

6.扩展问题

半包和粘包问题

如果上述步骤启动测试访问出现问题，则可能考虑一个点——粘包半包问题

【粘包半包】基本概念

使用TCP协议网络通讯，可能会出现粘包和半包问题，可以从实际例子理解：

【1】理想情况下，如果客户端连续2次发送消息，场景分析如下

# 客户端发送消息
// 第1次发送
hello noob!hello noob!hello noob!hello noob!
// 第2次发送
hello noob!hello noob!hello noob!hello noob!

# 服务端接收消息（可能存在情况：半包-每次收到的数据更少）
// 第1次接收
hello noob!hello noob!
// 第2次接收
hello noob!hello noob!hello noob!

# 服务端接收消息（可能存在情况：粘包-每次收到的数据更多）
// 第3次接收
hello noob!hello noob!hello noob!hello noob!hello noob!hello noob!

【2】实际案例分析

编写代码：VertxTcpClientTest、VertxTcpServerTest进行测试

/**
 * Vertx TCP 服务器
 */
@Slf4j
public class VertxTcpServerTest {

    public void doStart(int port) {
        // 创建 Vert.x 实例
        Vertx vertx = Vertx.vertx();

        // 创建 TCP 服务器
        NetServer server = vertx.createNetServer();

        // 处理请求
        server.connectHandler(socket->{
            // 处理连接
            socket.handler(buffer -> {
                /*
               // 处理接收到的字节数组
                byte[] requestData = buffer.getBytes();
                // 自定义字节数组处理逻辑（例如解析请求、调用服务、构造响应等）
                byte[] responseData = handleRequeset(requestData);
                // 发送响应(向连接到服务器的客户端发送数据,数据格式为Buffer（Vertx提供的字节数组缓冲区实现）)
                socket.write(Buffer.buffer(responseData));
                 */
                String testMessage = "hello noob!hello noob!hello noob!hello noob!";
                int messageLength = testMessage.getBytes().length;
                int bufferLength = buffer.getBytes().length;
                if(bufferLength<messageLength){
                    System.out.println("半包,length="+bufferLength);
                    return;
                }
                if(bufferLength>messageLength){
                    System.out.println("粘包,length="+bufferLength);
                    return;
                }
                String str = new String(buffer.getBytes(0, messageLength));
                System.out.println(str);
                if(testMessage.equals(str)){
                    System.out.println("数据接收正常");
                }
            });
        });

        // 启动 TCP 服务器并监听指定端口
        server.listen(port, result -> {
            if (result.succeeded()) {
                log.info("TCP server started on port " + port);
            } else {
                log.info("Failed to start TCP server: " + result.cause());
            }
        });
    }

    /**
     * 编写处理请求逻辑（结合实际业务场景编写）
     * @param requestData
     * @return
     */
    private byte[] handleRequeset(byte[] requestData) {
        return "hello Vertx Server".getBytes();
    }

    public static void main(String[] args) {
        new VertxTcpServerTest().doStart(8888);
    }
}

/**
 * Vertx TCP 请求客户端
 */
public class VertxTcpClientTest {

    public static void main(String[] args) {
        new VertxTcpClientTest().start();
    }

    /**
     * 发送请求
     */
    public void start(){
        // 创建Vert.x实例
        Vertx vertx = Vertx.vertx();

        vertx.createNetClient().connect(8888, "localhost", result -> {
            if (result.succeeded()) {
                System.err.println("connect to TCP server");
                io.vertx.core.net.NetSocket socket = result.result();

                // 发送数据(模拟发送1000次请求)
                for(int i=0;i<1000;i++){
                    socket.write("hello noob!hello noob!hello noob!hello noob!");
                }

                // 接收响应
                socket.handler(buffer -> {
                    System.out.println("received response from server: " + buffer.toString());
                });
            } else {
                System.err.println("fail to connect to TCP server");
            }
        });

    }
}

如果直接启动服务端，随后启动客户端会发现控制台没有打印信息（如果设置断点一步步放行则可看到对应的输出信息），socker编程启动，但是控制台不打印日志可能是有许多问题导致（日志框架、缓冲问题等），此处也可能是一个低级错误，例如本地启动服务端、客户端，如果启动客户端，对应socker监听应该要查看的是服务端的日志输出（即VertxTcpClientTest启动测试，应该对应查看请求响应服务的日志VertxTcpServerTest，不要混淆了概念）

如何解决半包？

解决半包的核心思路是：在消息头中设置请求体的长度,服务端接收时，判断每次消息的长度是否符合预期，不完整就不读，留到下一次接收到消息时再读取。

if(buffer==null||buffer.length()==0){
	throw new RuntimeException("消息buffer为空");
}
if(buffer.getBytes().length<ProtocolConstant.MESSAGE_HEADER_LENGTH){
	throw new RuntimeException("出现半包问题");
}

如何解决粘包？

解决粘包的核心思路：每次只读取指定长度的数据，超过长度的留着下一次接收到信息再读取

// 解决粘包问题，只读指定长度的数据
byte[] bodyBytes=buffer.getBytes(17,17+header.getBodyLength());

实现思路看上去简单，但是实现还是比较麻烦，要记录每次接收到的消息位置，维护字节数字缓存。因此考虑使用Vert.x解决半包和粘包问题

Vert.x解决半包和粘包

在Vert.x框架中，可以使用内置的RecordParser 完美解决半包粘包，它的作用是：保证下次读取到特定长度的字符（先不要急着直接修改业务代码，而是先学会该类库的使用，跑通测试流程，再引入到自己的业务代码中）

RecordParser读取固定长度消息

/**
 * Vertx TCP 服务器
 */
@Slf4j
public class VertxTcpServerTestByParse {

    public void doStart(int port) {
        // 创建 Vert.x 实例
        Vertx vertx = Vertx.vertx();

        // 创建 TCP 服务器
        NetServer server = vertx.createNetServer();

        // 处理请求
        server.connectHandler(socket->{
            // 处理连接
            socket.handler(buffer -> {
                String testMessage = "hello noob!hello noob!hello noob!hello noob!";
                int messageLength = testMessage.getBytes().length;

                // 构造parse
                RecordParser parser = RecordParser.newFixed(messageLength);
                parser.setOutput(new Handler<Buffer>() {
                    @Override
                    public void handle(Buffer buffer) {
                        String str = new String(buffer.getBytes());
                        System.out.println(str);
                        if(testMessage.equals(str)){
                            System.out.println("数据接收正常");
                        }
                    }
                });
                // 使用parse
                socket.handler(parser);
            });
        });

        // 启动 TCP 服务器并监听指定端口
        server.listen(port, result -> {
            if (result.succeeded()) {
                log.info("TCP server started on port " + port);
            } else {
                log.info("Failed to start TCP server: " + result.cause());
            }
        });
    }
}

实际运用中，消息体的长度是不固定的，所以要通过调整RecordParser的固定长度(变长)来解决。此处可以考虑将读取完整的消息拆分为2次：

【1】先完整读取请求头信息，由于请求头信息长度是固定的，可以使用RecordParser 保证每次都完整读取

【2】再根据请求头长度信息更改RecordParser 的固定长度，保证完整获取到请求体

/**
 * Vertx TCP 服务器
 */
@Slf4j
public class VertxTcpServerTestByParse {

    public void doStart(int port) {
        // 创建 Vert.x 实例
        Vertx vertx = Vertx.vertx();

        // 创建 TCP 服务器
        NetServer server = vertx.createNetServer();

        // 处理请求
        server.connectHandler(socket -> {
            // 处理连接
            socket.handler(buffer -> {
                // 构造parse
                RecordParser parser = RecordParser.newFixed(8);
                parser.setOutput(new Handler<Buffer>() {
                    // 初始化
                    int size = -1;
                    // 一次完整的读取（头+体）
                    Buffer resultBuffer = Buffer.buffer();

                    @Override
                    public void handle(Buffer buffer) {
                        if (-1 == size) {
                            // 读取消息体长度
                            size = buffer.getInt(4);
                            parser.fixedSizeMode(size);
                            // 写入头信息到结果
                            resultBuffer.appendBuffer(buffer);
                        } else {
                            // 写入体信息到结果
                            resultBuffer.appendBuffer(buffer);
                            System.out.println(resultBuffer.toString());
                            // 重置一轮
                            parser.fixedSizeMode(8);
                            size = -1;
                            resultBuffer = Buffer.buffer();
                        }
//                        System.out.println(resultBuffer);
                    }
                });
                // 使用parse
                socket.handler(parser);
            });
        });

        // 启动 TCP 服务器并监听指定端口
        server.listen(port, result -> {
            if (result.succeeded()) {
                log.info("TCP server started on port " + port);
            } else {
                log.info("Failed to start TCP server: " + result.cause());
            }
        });
    }

    /**
     * 编写处理请求逻辑（结合实际业务场景编写）
     * @param requestData
     * @return
     */
    private byte[] handleRequest(byte[] requestData) {
        return "hello Vertx Server".getBytes();
    }

    public static void main(String[] args) {
        new VertxTcpServerTestByParse().doStart(8888);
    }
}

/**
 * Vertx TCP 请求客户端
 */
public class VertxTcpClientTestByParse {

    public static void main(String[] args) {
        new VertxTcpClientTestByParse().start();
    }

    /**
     * 发送请求
     */
    public void start() {
        // 创建Vert.x实例
        Vertx vertx = Vertx.vertx();

        vertx.createNetClient().connect(8888, "localhost", result -> {
            if (result.succeeded()) {
                System.err.println("connect to TCP server");
                io.vertx.core.net.NetSocket socket = result.result();

                // 发送数据(模拟发送1000次请求)
                for (int i = 0; i < 1000; i++) {
                    Buffer buffer = Buffer.buffer();
                    String str = "hello noob!hello noob!hello noob!hello noob!";
                    buffer.appendInt(0);
                    buffer.appendInt(str.getBytes().length);
                    buffer.appendBytes(str.getBytes());
                    socket.write(buffer);
                }

                // 接收响应
                socket.handler(buffer -> {
                    System.out.println("received response from server: " + buffer.toString());
                });
            } else {
                System.err.println("fail to connect to TCP server");
            }
        });

    }
}

public class TcpServerHandler implements Handler<NetSocket> {
    /**
     * 处理请求
     * @param netSocket the event to handle
     */
    @Override
    public void handle(NetSocket netSocket) {
        TcpBufferHandlerWrapper bufferHandlerWrapper = new TcpBufferHandlerWrapper(buffer -> {
            // ------ 处理逻辑 ------
        });
        netSocket.handler(bufferHandlerWrapper);
    }
}

VertxTcpClient、ServiceProxy代码优化

此前将所有的发送请求、处理响应代码写到了ServiceProxy中，使得这个类看起来臃肿不堪，可以做个优化，将所有的请求响应逻辑提取出来，封装为单独的VertxTcpClient类

相当于将ServiceProxy中TCP处理响应的业务逻辑抽离出来，放到VertxTcpClient中

/**
 * Vertx TCP 请求客户端
 */
public class VertxTcpClient {
    /**
     * 发送请求封装
     *
     * @param rpcRequest
     * @param serviceMetaInfo
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static RpcResponse doRequest(RpcRequest rpcRequest, ServiceMetaInfo serviceMetaInfo) throws InterruptedException, ExecutionException {
        // 发送 TCP 请求
        Vertx vertx = Vertx.vertx();
        NetClient netClient = vertx.createNetClient();
        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        netClient.connect(serviceMetaInfo.getServicePort(), serviceMetaInfo.getServiceHost(),
                result -> {
                    if (!result.succeeded()) {
                        System.err.println("Failed to connect to TCP server");
                        return;
                    }
                    NetSocket socket = result.result();
                    // 发送数据，构造消息
                    ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                    ProtocolMessage.Header header = new ProtocolMessage.Header();
                    header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                    header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                    header.setSerializer((byte) ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer()).getKey());
                    header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                    // 生成全局请求 ID
                    header.setRequestId(IdUtil.getSnowflakeNextId());
                    protocolMessage.setHeader(header);
                    protocolMessage.setBody(rpcRequest);

                    // 编码请求
                    try {
                        Buffer encodeBuffer = ProtocolMessageEncoder.encode(protocolMessage);
                        socket.write(encodeBuffer);
                    } catch (IOException e) {
                        throw new RuntimeException("协议消息编码错误");
                    }

                    // 接收响应
                    TcpBufferHandlerWrapper bufferHandlerWrapper = new TcpBufferHandlerWrapper(
                            buffer -> {
                                try {
                                    ProtocolMessage<RpcResponse> rpcResponseProtocolMessage =
                                            (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                                    responseFuture.complete(rpcResponseProtocolMessage.getBody());
                                } catch (IOException e) {
                                    throw new RuntimeException("协议消息解码错误");
                                }
                            }
                    );
                    socket.handler(bufferHandlerWrapper);

                });
        RpcResponse rpcResponse = responseFuture.get();
        // 记得关闭连接
        netClient.close();
        return rpcResponse;
    }
}

简化ServiceProx服务调用：

RpcResponse rpcResponse = VertxTcpClient.doRequest(rpcRequest,selectedServiceMetaInfo);
return rpcResponse.getData();

启动测试：再次启动CoreProviderSample、EasyConsumerSample访问测试

扩展说明

扩展说明

（1）自己定义一个占用空间更少的RPC协议的消息结构

参考思路：序列化方式字段目前占用了8 bit,但其实总共就几种序列化方式，能否只占用4bit?其他字段也可以按照这种方式思考

思考问题：为什么tcpServer不提供个server接口，或者和httpServer共用接口?

替换这两个服务器(协议实现)涉及的改动点非常多，比如RPC协议、请求处理器等，不是直接能通过配置就替换的，而且RPC框架一般也不需要替换底层的协议,只使用TCP会更好

在系统设计时，按需设计、灵活应用，而不要无脑应用

昵称

邮箱

网址

按正序
按倒序
按热度

【noob-rpc】⑧扩展版RPC-自定义协议

预览: