Netty IM案例

wen java案例 2026-08-04 199

本文目录导读：

Netty IM案例

系统架构设计
自定义通信协议设计
服务端核心代码（im-server）
客户端实现（im-client）
测试与运行
进阶与优化方向
技术要点总结

下面为你提供一个基于 Netty 的即时通讯（IM）核心案例设计与代码实现，这个案例会涵盖客户端登录、单聊、群聊、心跳检测以及离线消息等核心功能。

系统架构设计

技术栈

Netty 4.x — 网络通信框架
Protobuf / JSON — 自定义协议（这里使用JSON简单演示，生产建议Protobuf）
Redis — 用户路由信息存储（保证分布式集群状态一致性）
MySQL + MyBatis — 好友关系与离线消息存储

核心模块划分

├── im-common          // 公共模块：协议、常量、工具类
├── im-server          // IM服务端
├── im-client          // IM客户端（用于测试）
└── im-admin           // 管理后台（可选）

自定义通信协议设计

由于Netty提供TCP传输,我们需要自定义消息边界，这里采用 长度域 + JSON体 方式：

// 消息结构：包体长度(4字节) + 消息类型(1字节) + 消息体(JSON)

协议定义（im-common模块）

public enum MsgType {
    LOGIN_REQ(1),       // 登录请求
    LOGIN_RESP(2),      // 登录响应
    CHAT_MSG(3),        // 聊天消息
    PING(4),            // 心跳
    PONG(5);            // 心跳响应
    public final int code;
    MsgType(int code) { this.code = code; }
    public static MsgType fromCode(int code) {
        for (MsgType type : values()) {
            if (type.code == code) return type;
        }
        return null;
    }
}

public class Message {
    private MsgType type;       // 消息类型
    private String body;        // JSON body
    private long timestamp;     // 客户端消息时间戳
    // getter/setter...
}

服务端核心代码（im-server）

Netty服务端启动器

public class IMServer {
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel serverChannel;
    public void start(int port) throws InterruptedException {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                 .channel(NioServerSocketChannel.class)
                 .option(ChannelOption.SO_BACKLOG, 1024)
                 .childOption(ChannelOption.SO_KEEPALIVE, true)
                 .childOption(ChannelOption.TCP_NODELAY, true)
                 .childHandler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel ch) {
                         ChannelPipeline pipeline = ch.pipeline();
                         // 拆包解包（处理半包/粘包）
                         pipeline.addLast(new LengthFieldBasedFrameDecoder(
                                 1024*1024, 0, 4, 1, 0));
                         pipeline.addLast(new JsonDecoder());
                         // 写数据处理
                         pipeline.addLast(new JsonEncoder());
                         // 业务逻辑
                         pipeline.addLast(new IMServerHandler());
                         // **空闲检测：5分钟内无读则关闭**
                         pipeline.addLast(new IdleStateHandler(300, 0, 0));
                         pipeline.addLast(new HeartbeatHandler());
                     }
                 });
        serverChannel = bootstrap.bind(port).sync().channel();
        System.out.println("IM Server started on port " + port);
    }
    public void shutdown() {
        if (serverChannel != null) serverChannel.close();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

自定义编解码器

// JSON编码器
public class JsonEncoder extends MessageToByteEncoder<Message> {
    private static final ObjectMapper mapper = new ObjectMapper();
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
        try {
            byte[] data = mapper.writeValueAsBytes(msg);
            out.writeInt(data.length);      // 长度
            out.writeByte(msg.getType().code); // 类型
            out.writeBytes(data);           // 数据
        } catch (JsonProcessingException e) {
            ctx.close();
        }
    }
}
// JSON解码器
public class JsonDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 因为前面已经使用LengthFieldBasedFrameDecoder拆包，这里直接解析
        int len = in.readInt();
        byte type = in.readByte();
        byte[] body = new byte[len];
        in.readBytes(body);
        Message msg = new Message();
        msg.setType(MsgType.fromCode(type));
        msg.setBody(new String(body, StandardCharsets.UTF_8));
        out.add(msg);
    }
}

核心业务处理器

public class IMServerHandler extends SimpleChannelInboundHandler<Message> {
    // 模拟在线用户表：userId -> Channel
    public static ConcurrentHashMap<Long, Channel> onlineUsers = new ConcurrentHashMap<>();
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 连接建立（尚未登录，此时不绑定用户）
        System.out.println("新连接: " + ctx.channel().remoteAddress());
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
        switch (msg.getType()) {
            case LOGIN_REQ:
                handleLogin(ctx, msg);
                break;
            case CHAT_MSG:
                handleChat(ctx, msg);
                break;
            case PING:
                // 响应PONG（心跳处理）
                Message pong = new Message();
                pong.setType(MsgType.PONG);
                ctx.writeAndFlush(pong);
                break;
            default:
                ctx.close();
        }
    }
    private void handleLogin(ChannelHandlerContext ctx, Message msg) {
        // 解析登录请求
        LoginRequest req = JSON.parseObject(msg.getBody(), LoginRequest.class);
        // 模拟认证成功
        // 1. 绑定用户ID与Channel
        Long userId = req.getUserId();
        onlineUsers.put(userId, ctx.channel());
        // 2. 返回登录响应
        Message resp = new Message();
        resp.setType(MsgType.LOGIN_RESP);
        resp.setBody("{\"code\":0,\"msg\":\"登录成功\"}");
        ctx.writeAndFlush(resp);
        System.out.println("用户 " + userId + " 登录成功");
    }
    private void handleChat(ChannelHandlerContext ctx, Message msg) {
        ChatRequest req = JSON.parseObject(msg.getBody(), ChatRequest.class);
        Long targetUserId = req.getTargetUserId();
        // 单聊
        if (req.getType() == 1) {
            Channel targetChannel = onlineUsers.get(targetUserId);
            if (targetChannel != null) {
                // 构建发给目标的消息（添加发送者信息）
                Message forward = new Message();
                forward.setType(MsgType.CHAT_MSG);
                forward.setBody(JSON.toJSONString(new ChatRequest(
                        req.getFromUserId(), req.getTargetUserId(), req.getContent())));
                targetChannel.writeAndFlush(forward);
            } else {
                // 用户不在线，存离线消息（可存入Redis/MySQL）
                saveOfflineMessage(req);
            }
        }
        // 群聊逻辑类似：遍历群成员，逐个发送
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 移除在线用户
        onlineUsers.entrySet().removeIf(entry -> entry.getValue() == ctx.channel());
        System.out.println("连接关闭");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

心跳与超时处理器

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IddStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                // 5分钟没收到消息，判定死亡连接
                System.out.println("心跳超时，关闭连接: " + ctx.channel().remoteAddress());
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

客户端实现（im-client）

public class IMClient {
    private Channel channel;
    private Long userId;
    public void connect(String host, int port) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                 .channel(NioSocketChannel.class)
                 .handler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel ch) {
                         ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024*1024, 0, 4, 1, 0));
                         ch.pipeline().addLast(new JsonDecoder());
                         ch.pipeline().addLast(new JsonEncoder());
                         ch.pipeline().addLast(new SimpleChannelInboundHandler<Message>() {
                             @Override
                             protected void channelRead0(ChannelHandlerContext, Message msg) {
                                 // 处理服务端推送的消息
                                 if (msg.getType() == MsgType.CHAT_MSG) {
                                     ChatRequest req = JSON.parseObject(msg.getBody(), ChatRequest.class);
                                     System.out.println("收到来自用户[" + req.getFromUserId() + "]: " + req.getContent());
                                 }
                             }
                         });
                     }
                 });
        channel = bootstrap.connect(host, port).sync().channel();
    }
    public void sendLogin(Long userId) {
        this.userId = userId;
        Message msg = new Message();
        msg.setType(MsgType.LOGIN_REQ);
        msg.setBody("{\"userId\":" + userId + "}");
        channel.writeAndFlush(msg);
    }
    public void sendChat(Long targetUserId, String content) {
        Message msg = new Message();
        msg.setType(MsgType.CHAT_MSG);
        msg.setBody(JSON.toJSONString(new ChatRequest(userId, targetUserId, content)));
        channel.writeAndFlush(msg);
    }
}

测试与运行

主测试类（两个客户端模拟单聊）

public class TestMain {
    public static void main(String[] args) throws Exception {
        // 启动服务端
        new Thread(() -> {
            try {
                new IMServer().start(8080);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        Thread.sleep(1000);
        // 客户端A登录
        IMClient clientA = new IMClient();
        clientA.connect("localhost", 8080);
        clientA.sendLogin(1001L);
        // 客户端B登录
        IMClient clientB = new IMClient();
        clientB.connect("localhost", 8080);
        clientB.sendLogin(1002L);
        Thread.sleep(500);
        // A给B发送消息
        clientA.sendChat(1002L, "你好，B！");
        Thread.sleep(2000);
    }
}

进阶与优化方向

协议优化：使用 Protobuf 代替JSON，性能提升数倍。
集群部署：接入 ZooKeeper 管理服务节点，通过 Redis Pub/Sub 实现跨节点消息转发。
消息可靠性：增加ACK确认机制，失败后自动重发。
离线消息：用户登录后拉取Redis/MySQL中的离线消息。
流量控制：结合Netty的高水位（WriteBufferWaterMark）实现背压。
安全：TLS/SSL加密，防止抓包。

技术要点总结

功能	实现方案
粘包/拆包	`LengthFieldBasedFrameDecoder` 长度域拆包
心跳检测	`IdleStateHandler` 定期检测读空闲
在线管理	`ConcurrentHashMap<userId, Channel>`
多端登录	使用Redis记录最新设备，踢出旧设备
消息有序	引入消息序号，配合Redis队列
性能调优	使用 `NioEventLoopGroup`，合理配置线程数