Netty

Netty 是 Java 后端开发中非常重要的高性能网络通信框架。尤其是你正在设计的 IM 系统，如果需要实现 WebSocket/TCP 长连接、心跳检测、消息编解码、连接管理和服务端推送，Netty 是值得深入掌握的技术。

一、Netty 是什么？

Netty 是一个基于 Java NIO 的异步事件驱动网络应用框架，可以用于构建 TCP、UDP、HTTP、WebSocket 等网络应用。

二、版本选择与 Maven 依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.2.19.Final</version>
</dependency>

三、Netty 核心架构

Netty 最重要的四个概念是：

1. Bootstrap

启动器。客户端使用 Bootstrap，服务端使用 ServerBootstrap。

2. EventLoopGroup

管理事件循环线程，负责连接接入、网络 I/O 与任务调度。

3. Channel

表示一个网络连接或监听通道，提供读写、关闭等操作。

4. ChannelPipeline

每个连接的处理链，负责解码、业务处理、编码等步骤。

1. 服务端线程模型

      ServerBootstrap
             |
      ServerSocketChannel
             |
     Boss EventLoopGroup
        接收连接
             |
    接入新的客户端 Channel
             |
    Worker EventLoopGroup
             |
   +---------+---------+
   |         |         |
Channel A  Channel B  Channel C
   |         |         |
Pipeline  Pipeline  Pipeline

BossGroup 负责接收新连接，WorkerGroup 负责已建立连接的 I/O。一个 EventLoop 通常可以处理多个 Channel，而一个已注册的 Channel 通常由固定的 EventLoop 负责处理其 I/O 事件。

最关键的原则：不要在 EventLoop 线程里执行耗时的阻塞操作。
例如，在 channelRead0() 中同步调用一个耗时的数据库查询，可能拖慢该 EventLoop 上其他连接的读写处理。

2. Pipeline 的执行顺序

假设 Pipeline 如下：

入站方向（Inbound）

Socket
  -> ByteBuf
  -> 解码器 Decoder
  -> 业务 Handler
  -> 后续 Handler

出站方向（Outbound）通常反向执行：

业务 Handler
  -> 编码器 Encoder
  -> ByteBuf
  -> Socket

入站事件通常从 Pipeline 的头部向尾部传播；出站操作会根据调用位置沿相反方向寻找对应的出站处理器。理解这个顺序，是排查编码、解码问题的基础。

四、实际例子

starter:

package com.easychat.websocket.netty;

import com.easychat.entity.config.Appconfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@Component("NettyWebsocketStart")
public class NettyWebsocketStart {

    private static final Logger logger = LoggerFactory.getLogger(NettyWebsocketStart.class);
    private EventLoopGroup bossGroup = new NioEventLoopGroup(1);

    private EventLoopGroup workGroup = new NioEventLoopGroup();

    /** 持有服务端 Channel，停机时先关闭，才能触发客户端断开并写完离线状态 */
    private Channel serverChannel;

    @Resource
    private Appconfig appconfig;

    @Resource
    private HandleWebSocket handleWebSocket;

    /**
     * 必须等 EventLoop 处理完 channelInactive（会更新数据库）之后再返回，
     * 否则 Spring 会先关闭 Hikari 连接池，断开连接时就会报 DataSource has been closed。
     */
    @PreDestroy
    public void close() {
        logger.info("正在关闭 Netty WebSocket 服务");
        if (serverChannel != null) {
            serverChannel.close().awaitUninterruptibly();
        }
        workGroup.shutdownGracefully().awaitUninterruptibly();
        bossGroup.shutdownGracefully().awaitUninterruptibly();
        logger.info("Netty 已关闭");
    }

    @Async
    public void startNetty() {
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workGroup);
            serverBootstrap.channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            // 设置几个重要的处理器
                            // 对http协议支持，使用http编码器，解码器
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(64 * 1024));
                            // 心跳
                            // readerIdleTime, writerIdleTime, allIdleTime, unit
                            pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
                            pipeline.addLast(new HandlerHeartBeat());

                            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
                            pipeline.addLast(handleWebSocket);
                        }
                    });

            ChannelFuture channelFuture = serverBootstrap.bind(appconfig.getWsport()).sync();
            serverChannel = channelFuture.channel();
            logger.info("netty服务器启动成功, 端口: {}", appconfig.getWsport());
            serverChannel.closeFuture().sync();
        } catch (InterruptedException e) {
            logger.warn("netty被打断");
        } catch (Exception e) {
            logger.error("启动netty失败", e);
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

HandleWebSocket:

package com.easychat.websocket.netty;

import com.easychat.entity.dto.TokenUserinfoDTO;
import com.easychat.entity.dto.MessageSendDto;
import com.easychat.entity.enums.MessageStatusEnum;
import com.easychat.entity.enums.MessageTypeEnum;
import com.easychat.entity.enums.UserContactStatusEnum;
import com.easychat.entity.po.ChatMessage;
import com.easychat.entity.po.ChatSessionUser;
import com.easychat.entity.po.UserContact;
import com.easychat.entity.po.UserInfo;
import com.easychat.entity.query.UserInfoQuery;
import com.easychat.exception.BusinessException;
import com.easychat.mappers.UserInfoMapper;
import com.easychat.redis.redisComponent;
import com.easychat.service.ChatMessageService;
import com.easychat.service.ChatSessionUserService;
import com.easychat.service.UserContactService;
import com.easychat.utils.JsonUtils;
import com.easychat.websocket.messageHandle;
import com.easychat.websocket.ChannelContextUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@ChannelHandler.Sharable
public class HandleWebSocket extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private static final Logger logger = LoggerFactory.getLogger(HandleWebSocket.class);

    @Resource
    private redisComponent redisComponent;

    @Resource
    private ChannelContextUtils channelContextUtils;

    @Resource
    private ChatMessageService chatMessageService;

    @Resource
    private ChatSessionUserService chatSessionUserService;

    @Resource
    private UserContactService userContactService;

    @Resource
    private UserInfoMapper<UserInfo, UserInfoQuery> userInfoMapper;

    @Resource
    private messageHandle messageHandle;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
        Channel channel = ctx.channel();
        Attribute<String> attribute = channel.attr(AttributeKey.valueOf(channel.id().toString()));
        String userId = attribute.get();
        redisComponent.setUserHeartBeat(userId);
        MessageSendDto<?> request = null;
        try {
            request = JsonUtils.convertJson2Obj(textWebSocketFrame.text(), MessageSendDto.class);
            if (request == null || !MessageTypeEnum.CHAT.getType().equals(request.getMessageType())) {
                throw new BusinessException("暂不支持该消息类型");
            }
            sendChatMessage(ctx, userId, request);
        } catch (BusinessException e) {
            logger.warn("WebSocket 消息校验失败，userId={}: {}", userId, e.getMessage());
            sendFailure(ctx, e.getMessage(), request == null ? null : request.getClientMessageId());
        } catch (Exception e) {
            logger.error("WebSocket 消息处理失败，userId={}", userId, e);
            sendFailure(ctx, "消息发送失败，请稍后重试", request == null ? null : request.getClientMessageId());
        }
    }

    private void sendChatMessage(ChannelHandlerContext ctx, String userId, MessageSendDto<?> request) {
        String contactId = request.getContactId();
        String content = request.getMessageContent() == null ? "" : request.getMessageContent().trim();
        if (userId == null || contactId == null || contactId.isEmpty() || content.isEmpty()) {
            throw new BusinessException("消息内容或联系人无效");
        }
        if (!Integer.valueOf(0).equals(request.getContactType())) {
            throw new BusinessException("当前只支持好友私聊");
        }
        if (content.length() > 10000) {
            throw new BusinessException("消息内容不能超过10000个字符");
        }

        ChatSessionUser senderSession = chatSessionUserService.getChatSessionUserByUserIdAndContactId(userId, contactId);
        if (senderSession == null || request.getSessionId() == null || !senderSession.getSessionId().equals(request.getSessionId())) {
            throw new BusinessException("无权向该会话发送消息");
        }
        UserContact contact = userContactService.getUserContactByUserIdAndContactId(userId, contactId);
        UserContact receiverContact = userContactService.getUserContactByUserIdAndContactId(contactId, userId);
        if (contact == null || receiverContact == null
                || !UserContactStatusEnum.FRIEND.getStatus().equals(contact.getStatus())
                || !UserContactStatusEnum.FRIEND.getStatus().equals(receiverContact.getStatus())) {
            throw new BusinessException("该联系人当前不可用");
        }
        ChatSessionUser receiverSession = chatSessionUserService.getChatSessionUserByUserIdAndContactId(contactId, userId);
        if (receiverSession == null || !senderSession.getSessionId().equals(receiverSession.getSessionId())) {
            throw new BusinessException("会话联系人信息不完整");
        }
        UserInfo sender = userInfoMapper.selectByUserId(userId);
        if (sender == null) {
            throw new BusinessException("发送用户不存在");
        }

        long sendTime = System.currentTimeMillis();
        ChatMessage chatMessage = ChatMessage.builder()
                .sessionId(senderSession.getSessionId())
                .messageType(MessageTypeEnum.CHAT.getType())
                .messageContent(content)
                .sendUserId(userId)
                .sendUserNickName(sender.getNickName())
                .sendTime(sendTime)
                .contactId(contactId)
                .contactType(0)
                .status(MessageStatusEnum.SENDED.getStatus())
                .build();
        chatMessageService.addAndUpdateSession(chatMessage);

        MessageSendDto<ChatSessionUser> receiverEvent = toChatEvent(chatMessage, receiverSession, request.getClientMessageId());
        receiverEvent.setContactId(contactId);

        MessageSendDto<ChatSessionUser> senderEvent = toChatEvent(chatMessage, senderSession, request.getClientMessageId());
        senderEvent.setContactId(senderSession.getContactId());
        senderEvent.setContactName(senderSession.getContactName());
        ctx.writeAndFlush(new TextWebSocketFrame(JsonUtils.convertObj2Json(senderEvent)));
        try {
            messageHandle.sendMsg(receiverEvent);
        } catch (Exception e) {
            // The message is committed and acknowledged to its sender. The recipient can still load it from history.
            logger.error("消息已保存，但通知接收方失败，messageId={}, contactId={}", chatMessage.getMessageId(), contactId, e);
        }
    }

    private MessageSendDto<ChatSessionUser> toChatEvent(ChatMessage message, ChatSessionUser session, String clientMessageId) {
        return MessageSendDto.<ChatSessionUser>builder()
                .messageId(message.getMessageId())
                .sessionId(message.getSessionId())
                .sendUserId(message.getSendUserId())
                .sendUserNickName(message.getSendUserNickName())
                .messageContent(message.getMessageContent())
                .messageType(message.getMessageType())
                .sendTime(message.getSendTime())
                .contactId(session.getContactId())
                .contactName(session.getContactName())
                .contactType(0)
                .status(message.getStatus())
                .clientMessageId(clientMessageId)
                .extendData(session)
                .build();
    }

    private void sendFailure(ChannelHandlerContext ctx, String reason, String clientMessageId) {
        MessageSendDto<Void> response = MessageSendDto.<Void>builder()
                .messageType(MessageTypeEnum.CHAT.getType())
                .messageContent(reason == null || reason.isEmpty() ? "消息发送失败" : reason)
                .status(2)
                .clientMessageId(clientMessageId)
                .build();
        ctx.writeAndFlush(new TextWebSocketFrame(JsonUtils.convertObj2Json(response)));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("有新的链接加入");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("有链接断开");
        channelContextUtils.removeContext(ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.warn("websocket连接异常: {}", cause.getMessage());
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
            WebSocketServerProtocolHandler.HandshakeComplete complete = (WebSocketServerProtocolHandler.HandshakeComplete) evt;
            String token = new QueryStringDecoder(complete.requestUri()).parameters()
                    .getOrDefault("token", java.util.Collections.<String>emptyList())
                    .stream()
                    .findFirst()
                    .orElse(null);
            if (token == null) {
                logger.warn("WebSocket 握手缺少 token 查询参数");
                ctx.channel().close();
                return;
            }

            TokenUserinfoDTO tokenUserInfoDTO = redisComponent.getTokenUserInfoDTO(token);
            if (null == tokenUserInfoDTO) {
               logger.warn("WebSocket 握手 token 无效");
               ctx.channel().close();
               return;
            }

            channelContextUtils.addContext(tokenUserInfoDTO.getUserId(), ctx.channel());
        }
    }
}

五、Handler 用法大全

Netty Handler 负责处理网络事件，是业务代码和网络框架之间的桥梁。

1. 常见回调

方法	触发时机
channelRegistered()	Channel 注册到 EventLoop
channelActive()	连接激活
channelRead()	收到入站数据
channelReadComplete()	一轮读取结束
channelWritabilityChanged()	Channel 可写状态变化
userEventTriggered()	收到用户事件，例如空闲事件
channelInactive()	连接断开
exceptionCaught()	传播到当前处理器的异常

在项目里，至少应该考虑连接建立、收到消息、连接关闭和异常处理这几个事件。

2. 两类最常用的入站处理器

ChannelInboundHandlerAdapter

直接接收 Object 类型消息，适合自定义复杂消息流程。

public class LoggingHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("收到消息: " + msg);

        // 把消息继续传给后续处理器
        ctx.fireChannelRead(msg);
    }
}

此处把 msg 传递给下一个处理器，相当于转移其后续处理责任。对于引用计数对象，不能在转交之后再随意释放它。

SimpleChannelInboundHandler<T>

只处理指定类型的消息：

public class StringHandler
        extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(
            ChannelHandlerContext ctx, String msg) {
        System.out.println(msg);
    }
}

它默认会在 channelRead0() 返回后自动释放已接收的引用计数消息。因此，如果要把 ByteBuf 保存到异步任务中继续使用，需要正确处理引用计数和对象生命周期。

3. ChannelHandlerContext 和 Channel

ctx.writeAndFlush(message);
ctx.channel().writeAndFlush(message);
ctx.close();

• ctx.writeAndFlush()：从当前 Handler 所在位置开始向前传播出站事件。
• channel.writeAndFlush()：通常从 Pipeline 的尾部开始传播出站事件。
• ctx.fireChannelRead(msg)：将入站事件传递给下一个 Handler。

如果某个编码器没有被调用，优先检查 Pipeline 中 Handler 的加入顺序，以及实际使用的是 ctx.writeAndFlush() 还是 channel.writeAndFlush()。

4. @Sharable 的含义

@ChannelHandler.Sharable
public class StatelessHandler extends ChannelInboundHandlerAdapter {
    // 不保存单个连接的可变状态
}

@Sharable 表示处理器可以被多个 Channel 共享，但只有在共享确实安全时才应使用。

不要把当前连接的用户 ID、业务请求缓存等可变状态直接存放在一个共享 Handler 的普通实例字段里。

六、心跳检测与空闲连接清理

IM 服务不能只依赖连接是否断开来判断客户端是否在线。网络切换、客户端休眠和中间代理超时都可能让连接进入异常状态。

Netty 提供 IdleStateHandler 检查读空闲、写空闲和读写均空闲的情况。

1. 服务端心跳 Handler

将下面处理器放在业务 Handler 前面：

pipeline.addLast(
    new IdleStateHandler(
        60,  // 60 秒没有读到数据，触发 READER_IDLE
        30,  // 30 秒没有写出数据，触发 WRITER_IDLE
        0,
        TimeUnit.SECONDS
    )
);

处理空闲事件：

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(
            ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;

            if (event.state() == IdleState.WRITER_IDLE) {
                // 示例协议：定期发送心跳
                ctx.writeAndFlush("PING\n");
            } else if (event.state() == IdleState.READER_IDLE) {
                // 长时间收不到客户端数据，关闭连接
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

需要配合客户端响应心跳，例如收到 PING 返回 PONG。心跳格式必须和实际协议保持一致；上面的代码只是用换行分隔的文本协议举例。

2. 心跳机制要注意什么？

TCP KeepAlive 不等于应用心跳。

TCP KeepAlive 主要由操作系统控制；业务层心跳则可以判断应用客户端是否仍能正常响应。

还要考虑移动端休眠、NAT 超时、负载均衡器连接超时，以及心跳流量和检测误判等问题。不要把读空闲事件机械地当成客户端已经离线：例如，某些客户端可能只接收推送，不定时发送业务消息。