Netty实现聊天室案例

本文目录导读：

1. 项目依赖 (pom.xml)
2. 消息协议类
3. 服务端实现
4. 客户端实现
5. 运行说明
6. 功能总结

我来为你提供一个完整的Netty聊天室案例,包含服务端和客户端的完整实现。

项目依赖 (pom.xml)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>netty-chat</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <!-- Netty -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.86.Final</version>
        </dependency>
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.30</version>
            <optional>true</optional>
        </dependency>
        <!-- Gson -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.10.1</version>
        </dependency>
    </dependencies>
</project>

消息协议类

package com.example.chat.protocol;
import lombok.Data;
/**
 * 聊天消息协议
 */
@Data
public class ChatMessage {
    // 消息类型
    public static final int TYPE_SYSTEM = 0;     // 系统消息
    public static final int TYPE_LOGIN = 1;      // 登录
    public static final int TYPE_LOGOUT = 2;     // 登出
    public static final int TYPE_CHAT = 3;       // 聊天消息
    public static final int TYPE_USER_LIST = 4;  // 在线用户列表
    private int type;           // 消息类型
    private String from;        // 发送者
    private String to;          // 接收者（null表示广播）
    private String content;     // 消息内容
    private String timestamp;   // 时间戳
    public ChatMessage() {}
    public ChatMessage(int type, String from, String content) {
        this.type = type;
        this.from = from;
        this.content = content;
        this.timestamp = String.valueOf(System.currentTimeMillis());
    }
    public static ChatMessage systemMessage(String content) {
        return new ChatMessage(TYPE_SYSTEM, "系统", content);
    }
    public static ChatMessage chatMessage(String from, String to, String content) {
        ChatMessage msg = new ChatMessage(TYPE_CHAT, from, content);
        msg.setTo(to);
        return msg;
    }
}

服务端实现

1 服务端主类

package com.example.chat.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import com.google.gson.Gson;
import com.example.chat.protocol.ChatMessage;
public class ChatServer {
    private final int port;
    private final Gson gson = new Gson();
    public ChatServer(int port) {
        this.port = port;
    }
    public void run() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                            .addLast("decoder", new StringDecoder(CharsetUtil.UTF_8))
                            .addLast("encoder", new StringEncoder(CharsetUtil.UTF_8))
                            .addLast("handler", new ChatServerHandler(gson));
                    }
                });
            System.out.println("Chat Server started on port " + port);
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 8080;
        new ChatServer(port).run();
    }
}

2 服务端处理器

package com.example.chat.server;
import com.google.gson.Gson;
import com.example.chat.protocol.ChatMessage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class ChatServerHandler extends SimpleChannelInboundHandler<String> {
    // 保存所有连接的Channel
    private static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    // 保存用户和Channel的对应关系
    private static final Map<String, Channel> USER_CHANNEL_MAP = new ConcurrentHashMap<>();
    // 保存所有在线用户
    private static final Map<String, String> ONLINE_USERS = new ConcurrentHashMap<>();
    private final Gson gson;
    public ChatServerHandler(Gson gson) {
        this.gson = gson;
    }
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        // 新Channel加入
        CHANNEL_GROUP.add(ctx.channel());
        System.out.println("[客户端] " + ctx.channel().remoteAddress() + " 加入连接");
    }
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        // Channel移除
        Channel channel = ctx.channel();
        CHANNEL_GROUP.remove(channel);
        // 从用户映射中移除
        USER_CHANNEL_MAP.entrySet().removeIf(entry -> entry.getValue() == channel);
        // 清除在线用户
        String username = getUsernameByChannel(channel);
        if (username != null) {
            ONLINE_USERS.remove(username);
            System.out.println("[客户端] " + username + " 断开连接");
            // 广播用户离线消息
            ChatMessage logoutMsg = ChatMessage.systemMessage(username + " 已离开聊天室");
            broadcast(logoutMsg);
            // 广播更新后的用户列表
            broadcastUserList();
        }
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        try {
            ChatMessage chatMessage = gson.fromJson(msg, ChatMessage.class);
            Channel channel = ctx.channel();
            switch (chatMessage.getType()) {
                case ChatMessage.TYPE_LOGIN:
                    handleLogin(channel, chatMessage);
                    break;
                case ChatMessage.TYPE_CHAT:
                    handleChat(chatMessage);
                    break;
                case ChatMessage.TYPE_LOGOUT:
                    handleLogout(channel, chatMessage);
                    break;
                default:
                    System.out.println("Unknown message type: " + chatMessage.getType());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 处理登录
     */
    private void handleLogin(Channel channel, ChatMessage msg) {
        String username = msg.getFrom();
        // 检查用户名是否已被占用
        if (ONLINE_USERS.containsKey(username)) {
            ChatMessage errorMsg = ChatMessage.systemMessage("用户名已被占用，请更换用户名");
            channel.writeAndFlush(gson.toJson(errorMsg));
            channel.close();
            return;
        }
        // 保存用户信息
        ONLINE_USERS.put(username, username);
        USER_CHANNEL_MAP.put(username, channel);
        // 发送登录成功通知给当前用户
        ChatMessage successMsg = ChatMessage.systemMessage("登录成功！欢迎 " + username + " 加入聊天室");
        channel.writeAndFlush(gson.toJson(successMsg));
        // 广播用户加入消息
        ChatMessage joinMsg = ChatMessage.systemMessage(username + " 加入了聊天室");
        broadcast(joinMsg);
        // 广播更新后的在线用户列表
        broadcastUserList();
        System.out.println("[用户登录] " + username);
    }
    /**
     * 处理聊天消息
     */
    private void handleChat(ChatMessage msg) {
        String to = msg.getTo();
        if (to != null && !to.isEmpty()) {
            // 私聊
            sendPrivateMessage(msg);
        } else {
            // 群发
            broadcast(msg);
        }
    }
    /**
     * 发送私聊消息
     */
    private void sendPrivateMessage(ChatMessage msg) {
        String to = msg.getTo();
        Channel targetChannel = USER_CHANNEL_MAP.get(to);
        if (targetChannel != null) {
            // 设置时间戳
            msg.setTimestamp(getCurrentTime());
            // 发送给接收者
            targetChannel.writeAndFlush(gson.toJson(msg));
            // 同时发送给发送者一份（回显）
            Channel fromChannel = USER_CHANNEL_MAP.get(msg.getFrom());
            if (fromChannel != null) {
                fromChannel.writeAndFlush(gson.toJson(msg));
            }
        } else {
            // 用户不在线
            Channel fromChannel = USER_CHANNEL_MAP.get(msg.getFrom());
            if (fromChannel != null) {
                ChatMessage errorMsg = ChatMessage.systemMessage("用户 " + to + " 不在线");
                fromChannel.writeAndFlush(gson.toJson(errorMsg));
            }
        }
    }
    /**
     * 处理登出
     */
    private void handleLogout(Channel channel, ChatMessage msg) {
        String username = msg.getFrom();
        CHANNEL_GROUP.remove(channel);
        USER_CHANNEL_MAP.remove(username);
        ONLINE_USERS.remove(username);
        // 广播用户离线消息
        ChatMessage logoutMsg = ChatMessage.systemMessage(username + " 已离开聊天室");
        broadcast(logoutMsg);
        // 广播更新后的用户列表
        broadcastUserList();
        channel.close();
    }
    /**
     * 广播消息给所有在线用户
     */
    private void broadcast(ChatMessage msg) {
        msg.setTimestamp(getCurrentTime());
        String json = gson.toJson(msg);
        for (Channel channel : CHANNEL_GROUP) {
            if (channel.isActive()) {
                channel.writeAndFlush(json);
            }
        }
    }
    /**
     * 广播在线用户列表
     */
    private void broadcastUserList() {
        ChatMessage msg = ChatMessage.systemMessage("在线用户列表");
        msg.setType(ChatMessage.TYPE_USER_LIST);
        msg.setContent(String.join(",", ONLINE_USERS.keySet()));
        msg.setTimestamp(getCurrentTime());
        String json = gson.toJson(msg);
        for (Channel channel : CHANNEL_GROUP) {
            if (channel.isActive()) {
                channel.writeAndFlush(json);
            }
        }
    }
    /**
     * 根据Channel获取用户名
     */
    private String getUsernameByChannel(Channel channel) {
        for (Map.Entry<String, Channel> entry : USER_CHANNEL_MAP.entrySet()) {
            if (entry.getValue() == channel) {
                return entry.getKey();
            }
        }
        return null;
    }
    /**
     * 获取当前时间
     */
    private String getCurrentTime() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return sdf.format(new Date());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("发生异常: " + cause.getMessage());
        ctx.close();
    }
}

客户端实现

1 客户端主类

package com.example.chat.client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import com.google.gson.Gson;
import com.example.chat.protocol.ChatMessage;
import java.io.BufferedReader;
import java.io.InputStreamReader;
public class ChatClient {
    private final String host;
    private final int port;
    private final String username;
    private final Gson gson = new Gson();
    public ChatClient(String host, int port, String username) {
        this.host = host;
        this.port = port;
        this.username = username;
    }
    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                            .addLast("decoder", new StringDecoder(CharsetUtil.UTF_8))
                            .addLast("encoder", new StringEncoder(CharsetUtil.UTF_8))
                            .addLast("handler", new ChatClientHandler(gson));
                    }
                });
            ChannelFuture future = bootstrap.connect(host, port).sync();
            Channel channel = future.channel();
            // 发送登录消息
            sendLogin(channel);
            // 启动读取控制台输入的线程
            startConsoleReader(channel);
            channel.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
    /**
     * 发送登录消息
     */
    private void sendLogin(Channel channel) {
        ChatMessage loginMsg = new ChatMessage(ChatMessage.TYPE_LOGIN, username, "登录");
        channel.writeAndFlush(gson.toJson(loginMsg));
    }
    /**
     * 启动控制台输入监听
     */
    private void startConsoleReader(Channel channel) {
        Thread thread = new Thread(() -> {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
                String line;
                System.out.println("聊天室已开启，输入 'quit' 退出，格式: /msg [用户名] [消息] 发送私聊");
                while ((line = reader.readLine()) != null) {
                    if (line.equalsIgnoreCase("quit")) {
                        sendLogout(channel);
                        break;
                    } else if (line.startsWith("/msg")) {
                        // 私聊命令
                        processPrivateCommand(channel, line);
                    } else {
                        // 群聊消息
                        Channel currentChannel = channel;
                        ChatMessage msg = ChatMessage.chatMessage(username, null, line);
                        currentChannel.writeAndFlush(gson.toJson(msg));
                    }
                }
                channel.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        thread.setDaemon(true);
        thread.start();
    }
    /**
     * 处理私聊命令
     */
    private void processPrivateCommand(Channel channel, String command) {
        String[] parts = command.split("\\s+", 3);
        if (parts.length >= 3) {
            String targetUser = parts[1];
            String content = parts[2];
            ChatMessage msg = ChatMessage.chatMessage(username, targetUser, content);
            channel.writeAndFlush(gson.toJson(msg));
        } else {
            System.out.println("使用格式: /msg [用户名] [消息]");
        }
    }
    /**
     * 发送登出消息
     */
    private void sendLogout(Channel channel) {
        ChatMessage logoutMsg = new ChatMessage(ChatMessage.TYPE_LOGOUT, username, "退出");
        channel.writeAndFlush(gson.toJson(logoutMsg));
    }
    public static void main(String[] args) {
        String host = args.length > 0 ? args[0] : "127.0.0.1";
        int port = args.length > 1 ? Integer.parseInt(args[1]) : 8080;
        String username = args.length > 2 ? args[2] : ("用户" + System.currentTimeMillis() % 1000);
        try {
            new ChatClient(host, port, username).start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

2 客户端处理器

package com.example.chat.client;
import com.google.gson.Gson;
import com.example.chat.protocol.ChatMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class ChatClientHandler extends SimpleChannelInboundHandler<String> {
    private final Gson gson;
    public ChatClientHandler(Gson gson) {
        this.gson = gson;
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        try {
            ChatMessage chatMessage = gson.fromJson(msg, ChatMessage.class);
            switch (chatMessage.getType()) {
                case ChatMessage.TYPE_SYSTEM:
                    handleSystemMessage(chatMessage);
                    break;
                case ChatMessage.TYPE_CHAT:
                    handleChatMessage(chatMessage);
                    break;
                case ChatMessage.TYPE_USER_LIST:
                    handleUserList(chatMessage);
                    break;
                default:
                    // 默认显示
                    System.out.println("收到消息: " + msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 处理系统消息
     */
    private void handleSystemMessage(ChatMessage msg) {
        System.out.println();
        System.out.println("【系统】" + msg.getContent());
        System.out.print("> ");
    }
    /**
     * 处理聊天消息
     */
    private void handleChatMessage(ChatMessage msg) {
        String from = msg.getFrom();
        String to = msg.getTo();
        String content = msg.getContent();
        if (to != null && !to.isEmpty()) {
            // 私聊消息
            System.out.println();
            System.out.println("【私聊】" + from + " -> " + to + ": " + content);
        } else {
            // 群聊消息
            System.out.println();
            System.out.println("【" + from + "】: " + content);
        }
        System.out.print("> ");
    }
    /**
     * 处理用户列表
     */
    private void handleUserList(ChatMessage msg) {
        String users = msg.getContent();
        System.out.println();
        System.out.println("【当前在线用户】: " + users);
        System.out.print("> ");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("连接异常: " + cause.getMessage());
        ctx.close();
    }
}

运行说明

启动服务端：

java com.example.chat.server.ChatServer 8080

启动客户端（开启多个终端）：

java com.example.chat.client.ChatClient localhost 8080 Alice
java com.example.chat.client.ChatClient localhost 8080 Bob
java com.example.chat.client.ChatClient localhost 8080 Charlie

使用说明：

1. 连接后自动登录，用户名不可重复
2. 直接输入内容发送群聊消息
3. 使用 /msg 用户名 消息 发送私聊
4. 输入 quit 退出聊天室

功能总结

这个聊天室实现了以下功能：

• ✅ 支持多客户端同时在线
• ✅ 群聊功能（广播消息）
• ✅ 私聊功能（指定用户发送）
• ✅ 在线用户列表显示
• ✅ 用户上线/下线通知
• ✅ 用户名重复检查
• ✅ 断开连接自动清理
• ✅ 格式化消息显示

可以继续扩展的功能：

• 消息持久化存储
• 消息重发机制
• 文件传输功能
• 消息过滤和敏感词检测
• 用户头像和表情
• 心跳检测和断线重连