1. 优化

1.1 序列化算法

序列化，反序列化主要用在消息正文的转换上

  • 序列化时，需要将 Java 对象变为要传输的数据（可以是 byte[]，或 json 等，最终都需要变成 byte[]）

  • 反序列化时，需要将传入的正文数据还原成 Java 对象，便于处理 ```java public interface Serializer {

    // 反序列化方法 T deserialize(Class clazz, byte[] bytes);

    // 序列化方法 byte[] serialize(T object);

}

  1. ```java
  2. public enum SerializerAlgorithm implements Serializer {
  4.     JAVA {
  5.         @Override
  6.         public <T> T deserialize(Class<T> clazz, byte[] bytes) {
  7.             try {
  8.                 ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
  9.                 return (T) new ObjectInputStream(byteArrayInputStream).readObject();
  10.             } catch (Exception e) {
  11.                 throw new RuntimeException("JAVA deserialize error", e);
  12.             }
  13.         }
  15.         @Override
  16.         public <T> byte[] serialize(T object) {
  17.             try {
  18.                 ByteArrayOutputStream buffer = new ByteArrayOutputStream();
  19.                 new ObjectOutputStream(buffer).writeObject(object);
  20.                 return buffer.toByteArray();
  21.             } catch (IOException e) {
  22.                 throw new RuntimeException("JAVA serializer error", e);
  23.             }
  24.         }
  25.     },
  27.     GSON {
  29.         private final Gson gson = new Gson();
  31.         @Override
  32.         public <T> T deserialize(Class<T> clazz, byte[] bytes) {
  33.             return gson.fromJson(new String(bytes, StandardCharsets.UTF_8), clazz);
  34.         }
  36.         @Override
  37.         public <T> byte[] serialize(T object) {
  38.             return gson.toJson(object).getBytes(StandardCharsets.UTF_8);
  39.         }
  40.     };
  42.     public static SerializerAlgorithm getByType(Byte type) {
  43.         return SerializerAlgorithm.values()[type];
  44.     }
  46. }
  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
  5.     @Override
  6.     protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
  7.         ByteBuf out = ctx.alloc().buffer();
  8.         // 1. 4 字节的魔数
  9.         out.writeBytes(new byte[]{1, 2, 3, 4});
  10.         // 2. 1 字节的版本,
  11.         out.writeByte(1);
  12.         // 3. 1 字节的序列化方式 jdk 0 , json 1
  13.         out.writeByte(Config.getSerializerAlgorithm().ordinal());
  14.         // 4. 1 字节的指令类型
  15.         out.writeByte(msg.getMessageType());
  16.         // 5. 4 个字节
  17.         out.writeInt(msg.getSequenceId());
  18.         // 无意义，对齐填充
  19.         out.writeByte(0xff);
  20.         // 6. 获取内容的字节数组
  21.         byte[] bytes = Config.getSerializerAlgorithm().serialize(msg);
  22.         log.info(new String(bytes));
  23.         // 7. 长度
  24.         out.writeInt(bytes.length);
  25.         // 8. 写入内容
  26.         out.writeBytes(bytes);
  27.         outList.add(out);
  28.     }
  30.     @Override
  31.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  32.         int magicNum = in.readInt();
  33.         byte version = in.readByte();
  34.         byte serializerType = in.readByte();
  35.         byte messageType = in.readByte();
  36.         int sequenceId = in.readInt();
  37.         in.readByte();
  38.         int length = in.readInt();
  39.         byte[] bytes = new byte[length];
  40.         in.readBytes(bytes, 0, length);
  41.         SerializerAlgorithm algorithm = SerializerAlgorithm.getByType(serializerType);
  42.         Class<? extends Message> messageClass = Message.getMessageClass(messageType);
  43.         Message message = algorithm.deserialize(messageClass, bytes);
  44.         log.info("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
  45.         log.info("{}", message);
  46.         out.add(message);
  47.     }
  49. }

1.2 参数调优

1.2.1 CONNECT_TIMEOUT_MILLIS

属于 SocketChannal 参数

用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常

SO_TIMEOUT 主要用在阻塞 IO，阻塞 IO 中 accept，read 等都是无限等待的，如果不希望永远阻塞，使用它调整超时时间

  1. @Slf4j
  2. public class TestConnectionTimeout {
  3.     public static void main(String[] args) {
  4.         NioEventLoopGroup group = new NioEventLoopGroup();
  5.         try {
  6.             Bootstrap bootstrap = new Bootstrap()
  7.                     .group(group)
  8.                     .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 300)
  9.                     .channel(NioSocketChannel.class)
  10.                     .handler(new LoggingHandler());
  11.             ChannelFuture future = bootstrap.connect("127.0.0.1", 8080);
  12.             future.sync().channel().closeFuture().sync(); // 断点1
  13.         } catch (Exception e) {
  14.             e.printStackTrace();
  15.             log.debug("timeout");
  16.         } finally {
  17.             group.shutdownGracefully();
  18.         }
  19.     }
  20. }

1.2.2 SO_BACKLOG

属于 ServerSocketChannal 参数
image.png

  1. 第一次握手，client 发送 SYN 到 server，状态修改为 SYN_SEND，server 收到，状态改变为 SYN_REVD，并将该请求放入 sync queue 队列
  2. 第二次握手，server 回复 SYN + ACK 给 client，client 收到，状态改变为 ESTABLISHED，并发送 ACK 给 server
  3. 第三次握手，server 收到 ACK，状态改变为 ESTABLISHED，将该请求从 sync queue 放入 accept queue

在 linux 2.2 之前，backlog 大小包括了两个队列的大小，在 2.2 之后，分别用下面两个参数来控制

  • sync queue - 半连接队列
    • 大小通过 /proc/sys/net/ipv4/tcp_max_syn_backlog 指定，在 syncookies 启用的情况下，逻辑上没有最大值限制，这个设置便被忽略
  • accept queue - 全连接队列
    • 其大小通过 /proc/sys/net/core/somaxconn 指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值
    • 如果 accpet queue 队列满了，server 将发送一个拒绝连接的错误信息到 client

netty 中可以通过 option(ChannelOption.SO_BACKLOG, 值) 来设置大小

1.2.3 TCP_NODELAY

属于 SocketChannal 参数

1.2.4 SO_SNDBUF & SO_RCVBUF

SO_SNDBUF 属于 SocketChannal 参数

SO_RCVBUF 既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）

1.2.5 ALLOCATOR

属于 SocketChannal 参数

用来分配 ByteBuf， ctx.alloc()

1.2.6 RCVBUF_ALLOCATOR

属于 SocketChannal 参数

控制 netty 接收缓冲区大小

负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator 决定

2. RPC 案例

  1. @Slf4j
  2. public class ClientRequestHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
  4.     public static final Map<Integer, Promise<Object>> PROMISES = new ConcurrentHashMap<>();
  6.     @Override
  7.     protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
  8.         // 注意这里需要从缓存中移除
  9.         Promise<Object> promise = PROMISES.remove(msg.getSequenceId());
  10.         if (promise != null) {
  11.             Object returnValue = msg.getReturnValue();
  12.             Throwable exceptionValue = msg.getExceptionValue();
  13.             if (exceptionValue != null) {
  14.                 promise.setFailure(exceptionValue);
  15.             } else {
  16.                 promise.setSuccess(returnValue);
  17.             }
  18.         }
  19.     }
  21. }
  1. @Slf4j
  2. public class RpcClient {
  4.     private static final Channel channel;
  6.     static {
  7.         channel = initChannel();
  8.     }
  10.     private static final AtomicInteger REQUEST_COUNT = new AtomicInteger(0);
  12.     @SuppressWarnings("all")
  13.     public static <T> T getProxyService(Class<T> serviceInterface) {
  14.         return (T) Proxy.newProxyInstance(
  15.                 serviceInterface.getClassLoader(),
  16.                 new Class[]{serviceInterface},
  17.                 new InvocationHandler() {
  18.                     @Override
  19.                     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  20.                         // 1. 构造 rpc 请求
  21.                         int sequenceId = REQUEST_COUNT.incrementAndGet();
  22.                         RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
  23.                                 sequenceId,
  24.                                 serviceInterface.getName(),
  25.                                 method.getName(),
  26.                                 method.getReturnType(),
  27.                                 method.getParameterTypes(),
  28.                                 args
  29.                         );
  30.                         // 2. 发送 rpc 请求
  31.                         channel.writeAndFlush(rpcRequestMessage);
  32.                         // 3. 准备一个空 Promise 对象，来接收结果, 指定 promise 对象异步接收结果线程
  33.                         DefaultPromise<Object> promise = new DefaultPromise<>(channel.eventLoop());
  34.                         ClientRequestHandler.PROMISES.put(sequenceId, promise);
  35.                         // 4. 阻塞知道有结果
  36.                         promise.await();
  37.                         // 5. 此时已经获取到结果, 解析
  38.                         if (promise.isSuccess()) {
  39.                             // 正常返回
  40.                             return promise.getNow();
  41.                         } else {
  42.                             // 包装异常
  43.                             throw new RuntimeException(promise.cause());
  44.                         }
  45.                     }
  46.                 });
  47.     }
  49.     private static Channel initChannel() {
  50.         NioEventLoopGroup group = new NioEventLoopGroup();
  51.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  52.         try {
  53.             Bootstrap bootstrap = new Bootstrap();
  54.             bootstrap.channel(NioSocketChannel.class);
  55.             bootstrap.group(group);
  56.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  57.                 @Override
  58.                 protected void initChannel(SocketChannel ch) throws Exception {
  59.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  60.                     ch.pipeline().addLast(MESSAGE_CODEC);
  61.                     // 业务处理
  62.                     ch.pipeline().addLast(new ClientRequestHandler());
  63.                     // 用来判断是不是[读空闲时间过长]，或[写空闲时间过长]
  64.                     // 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件
  65.                     ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
  66.                     // ChannelDuplexHandler 可以同时作为入站和出站处理器
  67.                     ch.pipeline().addLast(new ChannelDuplexHandler() {
  68.                         // 用来触发特殊事件
  69.                         @Override
  70.                         public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  71.                             IdleStateEvent event = (IdleStateEvent) evt;
  72.                             // 触发了写空闲事件
  73.                             if (event.state() == IdleState.WRITER_IDLE) {
  74.                                 ctx.writeAndFlush(new PingMessage());
  75.                             }
  76.                         }
  77.                     });
  78.                 }
  79.             });
  80.             // 这里需要 sync 同步确保连接建立
  81.             Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
  82.             // 这里需要异步监听 channel 关闭事件
  83.             channel.closeFuture().addListener(new ChannelFutureListener() {
  84.                 @Override
  85.                 public void operationComplete(ChannelFuture future) throws Exception {
  86.                     group.shutdownGracefully();
  87.                     if (!future.isSuccess()) {
  88.                         Throwable cause = future.cause();
  89.                         log.error("客户端 channel 关闭异常", cause);
  90.                     }
  91.                 }
  92.             });
  93.             return channel;
  94.         } catch (Exception e) {
  95.             log.error("client init error", e);
  96.         }
  97.         return null;
  98.     }
  100. }
  1. public class RpcClientTest {
  3.     public static void main(String[] args) {
  4.         HelloService helloService = RpcClient.getProxyService(HelloService.class);
  5.         System.out.println(helloService.echoName("张三"));
  6.         System.out.println(helloService.getAge());
  7.     }
  9. }