Netty实战与调优-阿里云开发者社区

Netty实战与调优

聊天室业务介绍

/**
 * 用户管理接口
 */
public interface UserService {

    /**
     * 登录
     * @param username 用户名
     * @param password 密码
     * @return 登录成功返回 true, 否则返回 false
     */
    boolean login(String username, String password);
}

/**
 * 会话管理接口
 */
public interface Session {

    /**
     * 绑定会话
     * @param channel 哪个 channel 要绑定会话
     * @param username 会话绑定用户
     */
    void bind(Channel channel, String username);

    /**
     * 解绑会话
     * @param channel 哪个 channel 要解绑会话
     */
    void unbind(Channel channel);

    /**
     * 获取属性
     * @param channel 哪个 channel
     * @param name 属性名
     * @return 属性值
     */
    Object getAttribute(Channel channel, String name);

    /**
     * 设置属性
     * @param channel 哪个 channel
     * @param name 属性名
     * @param value 属性值
     */
    void setAttribute(Channel channel, String name, Object value);

    /**
     * 根据用户名获取 channel
     * @param username 用户名
     * @return channel
     */
    Channel getChannel(String username);
}

/**
 * 聊天组会话管理接口
 */
public interface GroupSession {

    /**
     * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
     * @param name 组名
     * @param members 成员
     * @return 成功时返回组对象, 失败返回 null
     */
    Group createGroup(String name, Set<String> members);

    /**
     * 加入聊天组
     * @param name 组名
     * @param member 成员名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group joinMember(String name, String member);

    /**
     * 移除组成员
     * @param name 组名
     * @param member 成员名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group removeMember(String name, String member);

    /**
     * 移除聊天组
     * @param name 组名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group removeGroup(String name);

    /**
     * 获取组成员
     * @param name 组名
     * @return 成员集合, 没有成员会返回 empty set
     */
    Set<String> getMembers(String name);

    /**
     * 获取组成员的 channel 集合, 只有在线的 channel 才会返回
     * @param name 组名
     * @return 成员 channel 集合
     */
    List<Channel> getMembersChannel(String name);
}

注册事件处理器

@Slf4j
public class ChatServer {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        LoginRequestMessageHandler LOGIN_HANDLER = new LoginRequestMessageHandler();
        ChatRequestMessageHandler CHAT_HANDLER = new ChatRequestMessageHandler();
        GroupCreateRequestMessageHandler GROUP_CREATE_HANDLER = new GroupCreateRequestMessageHandler();
        GroupChatMessageHandler GROUP_CHAT_HANDLER = new GroupChatMessageHandler();
        QuitHandler QUIT_HANDLER = new QuitHandler();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    //空闲检测 arg1:读空闲 arg2:写空闲 arg3:读写空闲  单位:s  0代表不关注
                    ch.pipeline().addLast(new IdleStateHandler(5,0,0));
                    //处理空闲事件
                    ch.pipeline().addLast(new ChannelDuplexHandler(){
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            if (evt instanceof IdleStateEvent){
                                IdleStateEvent event = (IdleStateEvent) evt;
                                if (event.state().equals(IdleState.READER_IDLE)){
                                    log.info("读空闲超过5s!");
                                }
                            }
                        }
                    });
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(LOGIN_HANDLER);
                    ch.pipeline().addLast(CHAT_HANDLER);
                    ch.pipeline().addLast(GROUP_CREATE_HANDLER);
                    ch.pipeline().addLast(GROUP_CHAT_HANDLER);
                    ch.pipeline().addLast(QUIT_HANDLER);
                }
            });
            Channel channel = serverBootstrap.bind(8080).sync().channel();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

客户端

@Slf4j
public class ChatClient {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
        AtomicBoolean LOGIN = new AtomicBoolean(false);
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
//                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    //添加心跳机制
                    ch.pipeline().addLast(new IdleStateHandler(0,3,0));
                    ch.pipeline().addLast(new ChannelDuplexHandler(){
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            if (evt instanceof IdleStateEvent){
                                IdleStateEvent event = (IdleStateEvent) evt;
                                //超过3s没有写入数据则触发WRITER_IDLE事件
                                if (IdleState.WRITER_IDLE.equals(event.state())){
//                                    log.info("发送心跳数据包...");
                                    ctx.writeAndFlush(new PingMessage());
                                }
                            }
                        }
                    });
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            new Thread(() -> {
                                Scanner scanner = new Scanner(System.in);
                                System.out.print("username:");
                                String username = scanner.nextLine();
                                System.out.print("password:");
                                String password = scanner.nextLine();
                                LoginRequestMessage request = new LoginRequestMessage(username, password);
                                ctx.channel().writeAndFlush(request);
                                //阻塞
                                try {
                                    WAIT_FOR_LOGIN.await();
                                } catch (InterruptedException e) {
                                    throw new RuntimeException(e);
                                }
                                if (LOGIN.get()) {
                                    while (true) {
                                        System.out.println("==================================");
                                        System.out.println("send [username] [content]");
                                        System.out.println("gsend [group name] [content]");
                                        System.out.println("gcreate [group name] [m1,m2,m3...]");
                                        System.out.println("gmembers [group name]");
                                        System.out.println("gjoin [group name]");
                                        System.out.println("gquit [group name]");
                                        System.out.println("quit");
                                        System.out.println("==================================");
                                        String command = scanner.nextLine();
                                        if (command != null && command.length() > 0) {
                                            String[] commandList = command.split(" ");
                                            switch (commandList[0]) {
                                                case "send":
                                                    ChatRequestMessage message = new ChatRequestMessage(username, commandList[1], commandList[2]);
                                                    ctx.channel().writeAndFlush(message);
                                                    break;
                                                case "gsend":
                                                    GroupChatRequestMessage groupMessage = new GroupChatRequestMessage(username, commandList[1], commandList[2]);
                                                    ctx.channel().writeAndFlush(groupMessage);
                                                    break;
                                                case "gcreate":
                                                    String members = commandList[2];
                                                    if (members != null && members.length() > 0) {
                                                        String[] memberList = members.split(",");
                                                        Set<String> memberSet = Arrays.stream(memberList).collect(Collectors.toSet());
                                                        memberSet.add(username);
                                                        GroupCreateRequestMessage groupCreateRequestMessage = new GroupCreateRequestMessage(commandList[1], memberSet);
                                                        ctx.channel().writeAndFlush(groupCreateRequestMessage);
                                                    }
                                                    break;
                                                case "gmembers":
                                                    GroupMembersRequestMessage groupMembersRequestMessage = new GroupMembersRequestMessage(commandList[1]);
                                                    ctx.channel().writeAndFlush(groupMembersRequestMessage);
                                                    break;
                                                case "gjoin":
                                                    GroupJoinRequestMessage groupJoinRequestMessage = new GroupJoinRequestMessage(username, commandList[1]);
                                                    ctx.channel().writeAndFlush(groupJoinRequestMessage);
                                                    break;
                                                case "gquit":
                                                    GroupQuitRequestMessage groupQuitRequestMessage = new GroupQuitRequestMessage(username, commandList[1]);
                                                    ctx.channel().writeAndFlush(groupQuitRequestMessage);
                                                    break;
                                                case "quit":
                                                    ctx.channel().close();
                                                    return;
                                            }
                                        }
                                    }
                                } else {
                                    ctx.channel().close();
                                }
                            }).start();
                        }

                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            log.info("msg:{}", msg);
                            if (msg instanceof LoginResponseMessage response) {
                                LOGIN.set(response.isSuccess());
                                WAIT_FOR_LOGIN.countDown();
                            }
                        }
                    });
                }
            });
            Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
            channel.closeFuture().sync();
        } catch (Exception e) {
            log.error("client error", e);
        } finally {
            group.shutdownGracefully();
        }
    }
}

@ChannelHandler.Sharable
public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
        String username = msg.getUsername();
        String password = msg.getPassword();
        boolean login = UserServiceFactory.getUserService().login(username, password);
        if (login) {
            SessionFactory.getSession().bind(ctx.channel(), username);
        }
        ctx.channel().writeAndFlush(new LoginResponseMessage(login, login ? "登录成功!" : "登录失败,用户名或密码有误!"));
    }
}

单聊

@ChannelHandler.Sharable
public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ChatRequestMessage chatRequestMessage) throws Exception {
        String to = chatRequestMessage.getTo();
        String from = chatRequestMessage.getFrom();
        String content = chatRequestMessage.getContent();
        Channel channel = SessionFactory.getSession().getChannel(to);
        if (channel!=null){
            //转发消息
            channel.writeAndFlush(new ChatResponseMessage(from,content));
        }
        else {
            channelHandlerContext.writeAndFlush(new ChatResponseMessage(false,"发送失败,用户不在线!"));
        }
    }
}

创建群聊

@ChannelHandler.Sharable
@Slf4j
public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, GroupCreateRequestMessage groupCreateRequestMessage) throws Exception {
        Set<String> members = groupCreateRequestMessage.getMembers();
        String groupName = groupCreateRequestMessage.getGroupName();
        GroupSession session = GroupSessionFactory.getGroupSession();
        Group group = session.createGroup(groupName, members);
        GroupCreateResponseMessage message = null;
        if (group != null) {
            message = new GroupCreateResponseMessage(false, "创建群聊失败!");
        } else {
            message = new GroupCreateResponseMessage(true, "创建群聊成功!");
            List<Channel> channels = session.getMembersChannel(groupName);
            channels.forEach(channel -> {
                GroupJoinResponseMessage joinResponseMessage = new GroupJoinResponseMessage(true, "您已被拉入群聊:" + groupName);
                channel.writeAndFlush(joinResponseMessage);
            });
        }
        channelHandlerContext.writeAndFlush(message);
    }
}

发送群聊消息

@ChannelHandler.Sharable
public class GroupChatMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, GroupChatRequestMessage groupChatRequestMessage) throws Exception {
        String groupName = groupChatRequestMessage.getGroupName();
        String msg = groupChatRequestMessage.getContent();
        String from = groupChatRequestMessage.getFrom();
        List<Channel> channels = GroupSessionFactory.getGroupSession().getMembersChannel(groupName);
        channels.forEach(channel -> {
            channel.writeAndFlush(new GroupChatResponseMessage(from, msg));
        });
    }
}

退出

@ChannelHandler.Sharable
@Slf4j
public class QuitHandler extends ChannelInboundHandlerAdapter {
    //处理channel正常关闭事件
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //移除channel
        SessionFactory.getSession().unbind(ctx.channel());
        log.info("channel:{}已断开!",ctx.channel());
    }

    //处理channel异常关闭事件
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        SessionFactory.getSession().unbind(ctx.channel());
        log.info("channel:{}异常断开!",ctx.channel());
    }
}

空闲检测

连接假死

原因

网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源。
公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，就这么一直耗着
应用程序线程阻塞，无法进行数据读写

问题

假死的连接占用的资源不能自动释放
向假死的连接发送数据，得到的反馈是发送超时

服务器端解决

怎么判断客户端连接是否假死呢？如果能收到客户端数据，说明没有假死。因此策略就可以定为，每隔一段时间就检查这段时间内是否接收到客户端数据，没有就可以判定为连接假死

                    //空闲检测 arg1:读空闲 arg2:写空闲 arg3:读写空闲  单位:s  0代表不关注
                    ch.pipeline().addLast(new IdleStateHandler(5,0,0));
                    //处理空闲事件
                    ch.pipeline().addLast(new ChannelDuplexHandler(){
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            if (evt instanceof IdleStateEvent){
                                IdleStateEvent event = (IdleStateEvent) evt;
                                if (event.state().equals(IdleState.READER_IDLE)){
                                    log.info("读空闲超过5s!");
                                    //关闭channel
                                }
                            }
                        }
                    });

客户端定时心跳

客户端可以定时向服务器端发送数据，只要这个时间间隔小于服务器定义的空闲检测的时间间隔，那么就能防止前面提到的误判，客户端可以定义如下心跳处理器

                    //添加心跳机制
                    ch.pipeline().addLast(new IdleStateHandler(0,3,0));
                    ch.pipeline().addLast(new ChannelDuplexHandler(){
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            if (evt instanceof IdleStateEvent){
                                IdleStateEvent event = (IdleStateEvent) evt;
                                //超过3s没有写入数据则触发WRITER_IDLE事件
                                if (IdleState.WRITER_IDLE.equals(event.state())){
                                    log.info("发送心跳数据包...");
                                    ctx.writeAndFlush(new PingMessage());
                                }
                            }
                        }
                    });

优化

拓展序列化算法

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

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

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

public interface Serializer {

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

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

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

  enum Algorithm implements Serializer {
        Java {
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                try {
                    ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
                    Object object = in.readObject();
                    return (T) object;
                } catch (IOException | ClassNotFoundException e) {
                    throw new RuntimeException("SerializerAlgorithm.Java 反序列化错误", e);
                }
            }

            @Override
            public <T> byte[] serialize(T object) {
                try {
                    ByteArrayOutputStream out = new ByteArrayOutputStream();
                    new ObjectOutputStream(out).writeObject(object);
                    return out.toByteArray();
                } catch (IOException e) {
                    throw new RuntimeException("SerializerAlgorithm.Java 序列化错误", e);
                }
            }
        },

        // Json 实现(引入了 Gson 依赖)
        Json {
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                return new Gson().fromJson(new String(bytes, StandardCharsets.UTF_8), clazz);
            }

            @Override
            public <T> byte[] serialize(T object) {
                return new Gson().toJson(object).getBytes(StandardCharsets.UTF_8);
            }
        };

        // 需要从协议的字节中得到是哪种序列化算法
        public static Algorithm getByInt(int type) {
            Algorithm[] array = Algorithm.values();
            if (type < 0 || type > array.length - 1) {
                throw new IllegalArgumentException("Out of index!");
            }
            return array[type];
        }
    }

增加配置类和配置文件

public abstract class Config {
    static Properties properties;
    static {
        try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
            properties = new Properties();
            properties.load(in);
        } catch (IOException e) {
            throw new ExceptionInInitializerError(e);
        }
    }
    public static int getServerPort() {
        String value = properties.getProperty("server.port");
        if(value == null) {
            return 8080;
        } else {
            return Integer.parseInt(value);
        }
    }
    public static Serializer.Algorithm getSerializerAlgorithm() {
        String value = properties.getProperty("serializer.algorithm");
        if(value == null) {
            return Serializer.Algorithm.Java;
        } else {
            return Serializer.Algorithm.valueOf(value);
        }
    }
}

配置文件

serializer.algorithm=Json

修改编解码器

public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
        ByteBuf out = ctx.alloc().buffer();
        // 1. 4 字节的魔数
        out.writeBytes(new byte[]{1, 2, 3, 4});
        // 2. 1 字节的版本,
        out.writeByte(1);
        // 3. 1 字节的序列化方式 jdk 0 , json 1
        out.writeByte(0);
        // 4. 1 字节的指令类型
        out.writeByte(msg.getMessageType());
        // 5. 4 个字节
        out.writeInt(msg.getSequenceId());
        // 无意义，对齐填充
        out.writeByte(0xff);
        // 6. 获取内容的字节数组
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(msg);
        byte[] bytes = bos.toByteArray();
        // 7. 长度
        out.writeInt(bytes.length);
        // 8. 写入内容
        out.writeBytes(bytes);
        outList.add(out);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int magicNum = in.readInt();
        byte version = in.readByte();
        byte serializerType = in.readByte();
        byte messageType = in.readByte();
        int sequenceId = in.readInt();
        in.readByte();
        int length = in.readInt();
        byte[] bytes = new byte[length];
        in.readBytes(bytes, 0, length);
        ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
        Message message = (Message) ois.readObject();
        log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
        log.debug("{}", message);
        out.add(message);
    }
}

参数调优

CONNECT_TIMEOUT_MILLIS

属于 SocketChannal的参数
用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常
注意：Netty 中不要用成了SO_TIMEOUT 主要用在阻塞 IO，而 Netty 是非阻塞 IO

bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,300);// SocketChannel0.3s内未建立连接就抛出异常

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

                    // Schedule connect timeout.
                    int connectTimeoutMillis = config().getConnectTimeoutMillis();//获取超时时间
                    if (connectTimeoutMillis > 0) { 
                        //创建定时任务
                        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                            @Override
                            public void run() {  //如果超时后Promise依然存在,则向Promise标记异常
                                ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                                ConnectTimeoutException cause =
                                        new ConnectTimeoutException("connection timed out: " + remoteAddress);
                                if (connectPromise != null && connectPromise.tryFailure(cause)) {
                                    close(voidPromise());
                                }
                            }
                        }, connectTimeoutMillis, TimeUnit.MILLISECONDS);//在connectTimeoutMillis后开始执行任务
                    }

SO_BACKLOG

该参数是ServerSocketChannel的参数

三次握手与连接队列

简单易懂的解释

1.第一次握手时，因为客户端与服务器之间的连接还未完全建立，连接会被放入半连接队列中

2.当完成三次握手以后，连接会被放入全连接队列中

3.服务器处理Accept事件是在TCP三次握手，也就是建立连接之后。服务器会从全连接队列中获取连接并进行处理

详细的解释

第一次握手，client 发送 SYN 到 server，状态修改为 SYN_SEND，server 收到，状态改变为 SYN_REVD，并将该请求放入 sync queue 队列

第二次握手，server 回复 SYN + ACK 给 client，client 收到，状态改变为 ESTABLISHED，并发送 ACK 给 server

第三次握手，server 收到 ACK，状态改变为 ESTABLISHED，将该请求从 sync queue 放入 accept queue

Linux的backlog 参数

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

半连接队列 - sync queue

大小通过 /proc/sys/net/ipv4/tcp_max_syn_backlog 指定，在 syncookies 启用的情况下，逻辑上没有最大值限制，这个设置便被忽略

全连接队列 - accept queue

其大小通过 /proc/sys/net/core/somaxconn 指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值

如果 accept queue 队列满了，server 将发送一个拒绝连接的错误信息到 client

SO_BACKLOG作用

在Netty中，SO_BACKLOG主要用于设置全连接队列的大小。当处理Accept的速率小于连接建立的速率时，全连接队列中堆积的连接数大于SO_BACKLOG设置的值时，便会抛出异常

设置方式

// 设置全连接队列，大小为2
new ServerBootstrap().option(ChannelOption.SO_BACKLOG, 2);

SO_BACKLOG默认值

查询方法调用ServerSocketChannelImpl.bind

    @Override
    public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
        synchronized (stateLock) {
            ensureOpen();
            if (localAddress != null)
                throw new AlreadyBoundException();
            if (isUnixSocket()) {
                localAddress = unixBind(local, backlog);
            } else {
                localAddress = netBind(local, backlog);
            }
        }
        return this;
    }

在方法NioServerSocketChannel.doBind方法中调用了bind方法绑定backlog,此处调用了config.getBacklog方法

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

查询config实现类

private final class NioServerSocketChannelConfig extends DefaultServerSocketChannelConfig

在父类DefaultServerSocketChannelConfig中可以看到backlog被赋值

public class DefaultServerSocketChannelConfig extends DefaultChannelConfig implements              ServerSocketChannelConfig {
    protected final ServerSocket javaSocket;
    private volatile int backlog = NetUtil.SOMAXCONN;//为backlog赋值
}

具体赋值操作

SOMAXCONN = AccessController.doPrivileged(new PrivilegedAction<Integer>() {
    @Override
    public Integer run() {
        // Determine the default somaxconn (server socket backlog) value of the platform.
        // The known defaults:
        // - Windows NT Server 4.0+: 200
        // - Linux and Mac OS X: 128
        int somaxconn = PlatformDependent.isWindows() ? 200 : 128;
        File file = new File("/proc/sys/net/core/somaxconn");
        BufferedReader in = null;
        try {
            // file.exists() may throw a SecurityException if a SecurityManager is used, so execute it in the
            // try / catch block.
            // See https://github.com/netty/netty/issues/4936
            if (file.exists()) {
                in = new BufferedReader(new FileReader(file));
                // 将somaxconn设置为Linux配置文件中设置的值
                somaxconn = Integer.parseInt(in.readLine());
                if (logger.isDebugEnabled()) {
                    logger.debug("{}: {}", file, somaxconn);
                }
            } else {
                ...
            }
            ...
        }  
        // 返回backlog的值
        return somaxconn;
    }
}

backlog的值会根据操作系统的不同，来
选择不同的默认值
- Windows 200
- Linux/Mac OS 128
如果配置文件/proc/sys/net/core/somaxconn存在，会读取配置文件中的值，并将backlog的值设置为配置文件中指定的
- 注意：windows系统中并不存在该文件

ulimit -n

属于操作系统参数

最大文件限制数

在linux下一切皆文件，开启一个进程就是打开一个文件，控制ulimit大小作用等同于限制进程及其子进程的资源使用

使用ulimit -a 可以查看当前系统的所有限制值，使用ulimit -n 可以查看当前的最大打开文件数。

新装的 linux 默认只有1024，当作负载较大的服务器时，很容易遇到error: too many open files。因此，需要将其改大。

使用 ulimit -n 65535 可即时修改，但重启后就无效了。（注ulimit -SHn 65535 等效 ulimit -n 65535，-S指soft，-H指hard)

设置永久生效

可以修改配置文件/etc/profile
vi /etc/profile
加入一行
ulimit  -SHn  65536
保存退出。然后加载配置文件
source /etc/profile
再次查看ulimit
ulimit -n
第二种修改方法

在/etc/security/limits.conf最后增加如下两行记录
soft nofile 65535
hard nofile 65535

TCP_NODELAY

属于SocketChannal参数
因为Nagle算法，数据包会堆积到一定的数量后一起发送，这就可能导致数据的发送存在一定的延时
该参数默认为false,如果不希望的发送被延时，则需要将该值设置为true

在有些网络通信的场景下，要求低延迟，这样就需要我们设置一些TCP的链接属性

客户端设置

bootstap.option(ChannelOption.TCP_NODELAY, true);

服务器端是在worker的Channel端设置属性

boot.childOption(ChannelOption.TCP_NODELAY, true);

设置这样做好的好处就是禁用nagle算法

Nagle算法试图减少TCP包的数量和结构性开销, 将多个较小的包组合成较大的包进行发送.但这不是重点, 关键是这个算法受TCP延迟确认影响, 会导致相继两次向连接发送请求包,
读数据时会有一个最多达500毫秒的延时.
TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据)
Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块

SO_SNDBUF & SO_RCVBUF

SO_SNDBUF 属于SocketChannal参数
SO_RCVBUF既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）
该参数用于指定接收方与发送方的滑动窗口大小
一般不建议设置该参数，系统会自动设置大小

ALLOCATOR

属于SocketChannal参数
用来配置ByteBuf是池化还是非池化，是直接内存还是堆内存

使用

// 选择ALLOCATOR参数，设置SocketChannel中分配的ByteBuf类型
// 第二个参数需要传入一个ByteBufAllocator，用于指定生成的 ByteBuf 的类型
new ServerBootstrap().childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator());

ByteBufAllocator类型

池化并使用直接内存

// true表示使用直接内存
new PooledByteBufAllocator(true);

池化并使用堆内存

// false表示使用堆内存
new PooledByteBufAllocator(false);

非池化并使用直接内存

// ture表示使用直接内存
new UnpooledByteBufAllocator(true);

非池化并使用堆内存

// false表示使用堆内存
new UnpooledByteBufAllocator(false);

RCVBUF_ALLOCATOR

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

Netty实战与调优