由浅入深Netty代码调优

快来打我* 2024-03-17 09:13 99阅读 0赞

目录

    1. 优化
    • 1.1 扩展序列化算法
  • 2 参数调优
    • 2.1 CONNECT_TIMEOUT_MILLIS
    • 2.2 SO_BACKLOG
    • 2.3 ulimit -n
    • 2.4 TCP_NODELAY
    • 2.5 SO_SNDBUF & SO_RCVBUF
    • 2.6 ALLOCATOR
    • 2.7 RCVBUF_ALLOCATOR

1. 优化

在这里插入图片描述

1.1 扩展序列化算法

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

  • 序列化时，需要将 Java 对象变为要传输的数据（可以是 byte[]，或 json 等，最终都需要变成 byte[]）
  • 反序列化时，需要将传入的正文数据还原成 Java 对象，便于处理

目前的代码仅支持 Java 自带的序列化，反序列化机制，核心代码如下

  1. // 反序列化
  2. byte[] body = new byte[bodyLength];
  3. byteByf.readBytes(body);
  4. ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(body));
  5. Message message = (Message) in.readObject();
  6. message.setSequenceId(sequenceId);
  8. // 序列化
  9. ByteArrayOutputStream out = new ByteArrayOutputStream();
  10. new ObjectOutputStream(out).writeObject(message);
  11. byte[] bytes = out.toByteArray();

为了支持更多序列化算法，抽象一个 Serializer 接口

  1. public interface Serializer {
  4.     // 反序列化方法
  5.     <T> T deserialize(Class<T> clazz, byte[] bytes);
  7.     // 序列化方法
  8.     <T> byte[] serialize(T object);
  10. }

提供两个实现，我这里直接将实现加入了枚举类 Serializer.Algorithm 中

  1. enum SerializerAlgorithm implements Serializer {
  3.     // Java 实现
  4.     Java {
  6.         @Override
  7.         public <T> T deserialize(Class<T> clazz, byte[] bytes) {
  9.             try {
  11.                 ObjectInputStream in = 
  12.                     new ObjectInputStream(new ByteArrayInputStream(bytes));
  13.                 Object object = in.readObject();
  14.                 return (T) object;
  15.             } catch (IOException | ClassNotFoundException e) {
  17.                 throw new RuntimeException("SerializerAlgorithm.Java 反序列化错误", e);
  18.             }
  19.         }
  21.         @Override
  22.         public <T> byte[] serialize(T object) {
  24.             try {
  26.                 ByteArrayOutputStream out = new ByteArrayOutputStream();
  27.                 new ObjectOutputStream(out).writeObject(object);
  28.                 return out.toByteArray();
  29.             } catch (IOException e) {
  31.                 throw new RuntimeException("SerializerAlgorithm.Java 序列化错误", e);
  32.             }
  33.         }
  34.     }, 
  35.     // Json 实现(引入了 Gson 依赖)
  36.     Json {
  38.         @Override
  39.         public <T> T deserialize(Class<T> clazz, byte[] bytes) {
  41.             return new Gson().fromJson(new String(bytes, StandardCharsets.UTF_8), clazz);
  42.         }
  44.         @Override
  45.         public <T> byte[] serialize(T object) {
  47.             return new Gson().toJson(object).getBytes(StandardCharsets.UTF_8);
  48.         }
  49.     };
  51.     // 需要从协议的字节中得到是哪种序列化算法
  52.     public static SerializerAlgorithm getByInt(int type) {
  54.         SerializerAlgorithm[] array = SerializerAlgorithm.values();
  55.         if (type < 0 || type > array.length - 1) {
  57.             throw new IllegalArgumentException("超过 SerializerAlgorithm 范围");
  58.         }
  59.         return array[type];
  60.     }
  61. }

增加配置类和配置文件

  1. public abstract class Config {
  3.     static Properties properties;
  4.     static {
  6.         try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
  8.             properties = new Properties();
  9.             properties.load(in);
  10.         } catch (IOException e) {
  12.             throw new ExceptionInInitializerError(e);
  13.         }
  14.     }
  15.     public static int getServerPort() {
  17.         String value = properties.getProperty("server.port");
  18.         if(value == null) {
  20.             return 8080;
  21.         } else {
  23.             return Integer.parseInt(value);
  24.         }
  25.     }
  26.     public static Serializer.Algorithm getSerializerAlgorithm() {
  28.         String value = properties.getProperty("serializer.algorithm");
  29.         if(value == null) {
  31.             return Serializer.Algorithm.Java;
  32.         } else {
  34.             return Serializer.Algorithm.valueOf(value);
  35.         }
  36.     }
  37. }

配置文件

  1. serializer.algorithm=Json

修改编解码器

  1. /**
  2.  * 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的
  3.  */
  4. public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
  6.     @Override
  7.     public void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
  9.         ByteBuf out = ctx.alloc().buffer();
  10.         // 1. 4 字节的魔数
  11.         out.writeBytes(new byte[]{
  12.     1, 2, 3, 4});
  13.         // 2. 1 字节的版本,
  14.         out.writeByte(1);
  15.         // 3. 1 字节的序列化方式 jdk 0 , json 1
  16.         out.writeByte(Config.getSerializerAlgorithm().ordinal());
  17.         // 4. 1 字节的指令类型
  18.         out.writeByte(msg.getMessageType());
  19.         // 5. 4 个字节
  20.         out.writeInt(msg.getSequenceId());
  21.         // 无意义，对齐填充
  22.         out.writeByte(0xff);
  23.         // 6. 获取内容的字节数组
  24.         byte[] bytes = Config.getSerializerAlgorithm().serialize(msg);
  25.         // 7. 长度
  26.         out.writeInt(bytes.length);
  27.         // 8. 写入内容
  28.         out.writeBytes(bytes);
  29.         outList.add(out);
  30.     }
  32.     @Override
  33.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  35.         int magicNum = in.readInt();
  36.         byte version = in.readByte();
  37.         byte serializerAlgorithm = in.readByte(); // 0 或 1
  38.         byte messageType = in.readByte(); // 0,1,2...
  39.         int sequenceId = in.readInt();
  40.         in.readByte();
  41.         int length = in.readInt();
  42.         byte[] bytes = new byte[length];
  43.         in.readBytes(bytes, 0, length);
  45.         // 找到反序列化算法
  46.         Serializer.Algorithm algorithm = Serializer.Algorithm.values()[serializerAlgorithm];
  47.         // 确定具体消息类型
  48.         Class<? extends Message> messageClass = Message.getMessageClass(messageType);
  49.         Message message = algorithm.deserialize(messageClass, bytes);
  50. //        log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
  51. //        log.debug("{}", message);
  52.         out.add(message);
  53.     }
  54. }

其中确定具体消息类型，可以根据 消息类型字节 获取到对应的 消息 class

  1. @Data
  2. public abstract class Message implements Serializable {
  5.     /**
  6.      * 根据消息类型字节，获得对应的消息 class
  7.      * @param messageType 消息类型字节
  8.      * @return 消息 class
  9.      */
  10.     public static Class<? extends Message> getMessageClass(int messageType) {
  12.         return messageClasses.get(messageType);
  13.     }
  15.     private int sequenceId;
  17.     private int messageType;
  19.     public abstract int getMessageType();
  21.     public static final int LoginRequestMessage = 0;
  22.     public static final int LoginResponseMessage = 1;
  23.     public static final int ChatRequestMessage = 2;
  24.     public static final int ChatResponseMessage = 3;
  25.     public static final int GroupCreateRequestMessage = 4;
  26.     public static final int GroupCreateResponseMessage = 5;
  27.     public static final int GroupJoinRequestMessage = 6;
  28.     public static final int GroupJoinResponseMessage = 7;
  29.     public static final int GroupQuitRequestMessage = 8;
  30.     public static final int GroupQuitResponseMessage = 9;
  31.     public static final int GroupChatRequestMessage = 10;
  32.     public static final int GroupChatResponseMessage = 11;
  33.     public static final int GroupMembersRequestMessage = 12;
  34.     public static final int GroupMembersResponseMessage = 13;
  35.     public static final int PingMessage = 14;
  36.     public static final int PongMessage = 15;
  37.     private static final Map<Integer, Class<? extends Message>> messageClasses = new HashMap<>();
  39.     static {
  41.         messageClasses.put(LoginRequestMessage, LoginRequestMessage.class);
  42.         messageClasses.put(LoginResponseMessage, LoginResponseMessage.class);
  43.         messageClasses.put(ChatRequestMessage, ChatRequestMessage.class);
  44.         messageClasses.put(ChatResponseMessage, ChatResponseMessage.class);
  45.         messageClasses.put(GroupCreateRequestMessage, GroupCreateRequestMessage.class);
  46.         messageClasses.put(GroupCreateResponseMessage, GroupCreateResponseMessage.class);
  47.         messageClasses.put(GroupJoinRequestMessage, GroupJoinRequestMessage.class);
  48.         messageClasses.put(GroupJoinResponseMessage, GroupJoinResponseMessage.class);
  49.         messageClasses.put(GroupQuitRequestMessage, GroupQuitRequestMessage.class);
  50.         messageClasses.put(GroupQuitResponseMessage, GroupQuitResponseMessage.class);
  51.         messageClasses.put(GroupChatRequestMessage, GroupChatRequestMessage.class);
  52.         messageClasses.put(GroupChatResponseMessage, GroupChatResponseMessage.class);
  53.         messageClasses.put(GroupMembersRequestMessage, GroupMembersRequestMessage.class);
  54.         messageClasses.put(GroupMembersResponseMessage, GroupMembersResponseMessage.class);
  55.     }
  56. }

2 参数调优

2.1 CONNECT_TIMEOUT_MILLIS

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

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

    @Slf4j
    public class TestConnectionTimeout {

    1. public static void main(String[] args) {
    3.     NioEventLoopGroup group = new NioEventLoopGroup();
    4.     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) {
    15.         e.printStackTrace();
    16.         log.debug("timeout");
    17.     } finally {
    19.         group.shutdownGracefully();
    20.     }
    21. }

    }

另外源码部分 io.netty.channel.nio.AbstractNioChannel.AbstractNioUnsafe#connect

  1. @Override
  2. public final void connect(
  3.         final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
  5.     // ...
  6.     // Schedule connect timeout.
  7.     int connectTimeoutMillis = config().getConnectTimeoutMillis();
  8.     if (connectTimeoutMillis > 0) {
  10.         connectTimeoutFuture = eventLoop().schedule(new Runnable() {
  12.             @Override
  13.             public void run() {
  15.                 ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
  16.                 ConnectTimeoutException cause =
  17.                     new ConnectTimeoutException("connection timed out: " + remoteAddress); // 断点2
  18.                 if (connectPromise != null && connectPromise.tryFailure(cause)) {
  20.                     close(voidPromise());
  21.                 }
  22.             }
  23.         }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
  24.     }
  25.     // ...
  26. }

2.2 SO_BACKLOG

  • 属于 ServerSocketChannal 参数

clientserversyns queueaccept queuebind()listen()connect()1. SYNSYN_SENDputSYN_RCVD2. SYN + ACKESTABLISHED3. ACKputESTABLISHEDaccept()clientserversyns queueaccept queue

  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. public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
  2.                                               implements ServerSocketChannelConfig {
  5.     private volatile int backlog = NetUtil.SOMAXCONN;
  6.     // ...
  7. }

课堂调试关键断点为：io.netty.channel.nio.NioEventLoop#processSelectedKey

oio 中更容易说明，不用 debug 模式

  1. public class Server {
  3.     public static void main(String[] args) throws IOException {
  5.         ServerSocket ss = new ServerSocket(8888, 2);
  6.         Socket accept = ss.accept();
  7.         System.out.println(accept);
  8.         System.in.read();
  9.     }
  10. }

客户端启动 4 个

  1. public class Client {
  3.     public static void main(String[] args) throws IOException {
  5.         try {
  7.             Socket s = new Socket();
  8.             System.out.println(new Date()+" connecting...");
  9.             s.connect(new InetSocketAddress("localhost", 8888),1000);
  10.             System.out.println(new Date()+" connected...");
  11.             s.getOutputStream().write(1);
  12.             System.in.read();
  13.         } catch (IOException e) {
  15.             System.out.println(new Date()+" connecting timeout...");
  16.             e.printStackTrace();
  17.         }
  18.     }
  19. }

第 1，2，3 个客户端都打印，但除了第一个处于 accpet 外，其它两个都处于 accept queue 中

  1. Tue Apr 21 20:30:28 CST 2020 connecting...
  2. Tue Apr 21 20:30:28 CST 2020 connected...

第 4 个客户端连接时

  1. Tue Apr 21 20:53:58 CST 2020 connecting...
  2. Tue Apr 21 20:53:59 CST 2020 connecting timeout...
  3. java.net.SocketTimeoutException: connect timed out

2.3 ulimit -n

  • 属于操作系统参数

2.4 TCP_NODELAY

  • 属于 SocketChannal 参数

2.5 SO_SNDBUF & SO_RCVBUF

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

2.6 ALLOCATOR

  • 属于 SocketChannal 参数
  • 用来分配 ByteBuf， ctx.alloc()

2.7 RCVBUF_ALLOCATOR

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

发表评论

表情：
评论列表 （有 0 条评论，99人围观）

还没有评论，来说两句吧...

相关阅读

    相关 代码调优

    “”代码调优“”首先确定程序中开销较大的部分，然后对齐修改，但其也是双刃剑，当可能存在的危害影响较大时，请考虑适当将效率放一放。接下来举几个常见的手段： （1）使用开销小的

    悠悠悠悠/ 0 赞/ 205 阅读

    相关 spark调优 JVM调优

    我们的堆内存分为：新生代，和年老代， 年轻代又分为：Eden区，幸存一区，幸存二区， 每一次访对象的时候，都是放入eden区域，和其中的一个幸存一区中，幸存二区是不放对象的

    ￡神魔★判官ぃ￡神魔★判官ぃ/ 0 赞/ 1678 阅读