用Netty实现WebSocket网络聊天室

用Netty实现WebSocket网络聊天室 | XINDOO

最近学习Netty，学习到WebSocket一章节，将Netty中WebSocket的样例代码做了一些简单的改造，实现了一个简易的WebSocket网络聊天室，源码并非完全自己实现，只是将一些别人的代码做了下整合改造，今分享至此，希望对大家学习Netty有所帮助。
首先介绍下什么是WebSocket，这就不得不先提到HTTP协议了。众所周知，在HTTP/2发布前，所有HTTP请求都是 请求-应答的 模式，这就意味着客户端只能向服务器要数据，然后服务端被动应答，而服务器无法主动将数据推送给客户端。这就导致一些高时效性的场景用HTTP就会有些问题，就拿实时聊天举例吧，客户端想知道近期有没有人说过话，就只能不断问服务器有没有人发了消息？有的话服务器就返回，没有就不返回，这种行为被称为轮询。轮询的问题在于如果询问的时间间隔太长，消息的及时性无法得到保证，但如果时间太短，对服务器的压力就会大幅提升（因为不断要请求响应）。有没有可能服务器有消息的时候，主动推送给客户端？
WebSocket因此而诞生，它允许客户端和服务端之间在HTTP之上建立一个全双工的TCP长连接，这里的关键点在于全双工，意味着服务端也能通过这个连接给客户端发送即时消息，从而解决了轮询的性能和时效性矛盾的问题。了解过Socket编程的同学应该很容易理解了，WebSocket其实本质上就是Socket，只不过WebSocket是建立在HTTP协议之上的。
回到我们的正题，如何用Netty+WebSocket写一个网络聊天室？其实Netty里已经封装好了HTTP和WebSocket的实现，我们只需要实现部分聊天室的功能即可，接下来看下我实现的完整代码：
首先是ServerBootstrap的部分，这里是Netty的启动入口。

复制

@Service
public class WebSocketServer {

    static final String WEBSOCKET_PATH = "/ws";

    private ChannelFuture f;

    @Resource
    private WebSocketFrameHandler webSocketFrameHandler;

    @PostConstruct
    private void init() {
        bind(Constant.SOCKET_PORT);
    }

    public static voud bind(int port) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new HttpServerCodec());  // netty中http协议的编解码
                            pipeline.addLast(new HttpObjectAggregator(65536));  
                            pipeline.addLast(new WebSocketServerCompressionHandler());
                            pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
                            pipeline.addLast(new WebSocketIndexPageHandler(WEBSOCKET_PATH));  // demo页面的handler，这个是非必须的，可以换成其他第三方的WebSocket客户端工具  
                            pipeline.addLast(webSocketFrameHandler); // 聊天室的主要逻辑
                        }
                    });
            Channel f = b.bind(port).sync().channel();
            f.closeFuture().sync();
        } catch (Exception e) {

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

因为HttpServerCodec HttpObjectAggregator WebSocketServerCompressionHandler WebSocketServerProtocolHandler是Netty组件中提供的组件，其作用就是完成Http和WebSocket协议数据到Java对象的相互转换，这里就不再展开了，我们直接看下剩下的两个Handler。
首先是WebSocketIndexPageHandler，这个也是我直接从Netty样例中Copy出来的，它的作用就是构建一个Http首页，这个首页实现了一个简单的WebSocket网页客户端，如果你不需要这个网页客户端，你也可直接删掉。

复制

public class WebSocketIndexPageHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final String websocketPath;

    public WebSocketIndexPageHandler(String websocketPath) {
        this.websocketPath = websocketPath;
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        // Generate an error page if response getStatus code is not OK (200).
        HttpResponseStatus responseStatus = res.status();
        if (responseStatus.code() != 200) {
            ByteBufUtil.writeUtf8(res.content(), responseStatus.toString());
            HttpUtil.setContentLength(res, res.content().readableBytes());
        }
        // Send the response and close the connection if necessary.
        boolean keepAlive = HttpUtil.isKeepAlive(req) && responseStatus.code() == 200;
        HttpUtil.setKeepAlive(res, keepAlive);
        ChannelFuture future = ctx.writeAndFlush(res);
        if (!keepAlive) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

    private static String getWebSocketLocation(ChannelPipeline cp, HttpRequest req, String path) {
        String protocol = "ws";
        if (cp.get(SslHandler.class) != null) {
            // SSL in use so use Secure WebSockets
            protocol = "wss";
        }
        return protocol + "://" + req.headers().get(HttpHeaderNames.HOST) + path;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
        // Handle a bad request.
        if (!req.decoderResult().isSuccess()) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST,
                                                                   ctx.alloc().buffer(0)));
            return;
        }

        // Allow only GET methods.
        if (!GET.equals(req.method())) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(req.protocolVersion(), FORBIDDEN,
                                                                   ctx.alloc().buffer(0)));
            return;
        }

        // Send the index page
        if ("/".equals(req.uri()) || "/index.html".equals(req.uri())) {
            String webSocketLocation = getWebSocketLocation(ctx.pipeline(), req, websocketPath);
            ByteBuf content = WebSocketServerIndexPage.getContent(webSocketLocation);
            FullHttpResponse res = new DefaultFullHttpResponse(req.protocolVersion(), OK, content);

            res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
            HttpUtil.setContentLength(res, content.readableBytes());

            sendHttpResponse(ctx, req, res);
        } else {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(req.protocolVersion(), NOT_FOUND,
                                                                   ctx.alloc().buffer(0)));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

上面这部分代码还依赖于一个静态的页面，我对这个页面做了简单的改造，方便大家愉快地一起聊天。

复制


public final class WebSocketServerIndexPage {

    public static ByteBuf getContent(String webSocketLocation) {
        return Unpooled.copiedBuffer("\n"
                                     + "<html><head><title>Web Socket Test</title></head>\n"
                                     + "<body>\n"
                                     + "<script type=\"text/javascript\">\n"
                                     + "var socket;\n"
                                     + "if (!window.WebSocket) {\n"
                                     + "  window.WebSocket = window.MozWebSocket;\n"
                                     + "}\n"
                                     + "if (window.WebSocket) {\n"
                                     + "  socket = new WebSocket(\"" + webSocketLocation + "\");\n"
                                     + "  socket.onmessage = function(event) {\n"
                                     + "    var ta = document.getElementById('responseText');\n"
                                     + "    ta.value = ta.value + '\\n' + event.data\n"
                                     + "    ta.scrollTop = ta.scrollHeight\n"
                                     + "  };\n"
                                     + "  socket.onopen = function(event) {\n"
                                     + "    var ta = document.getElementById('responseText');\n"
                                     + "    ta.value = \"Web Socket opened!\";\n"
                                     + "    ta.scrollTop = ta.scrollHeight\n"
                                     + "  };\n"
                                     + "  socket.onclose = function(event) {\n"
                                     + "    var ta = document.getElementById('responseText');\n"
                                     + "    ta.value = ta.value + \"Web Socket closed!\";\n"
                                     + "    ta.scrollTop = ta.scrollHeight\n"
                                     + "  };\n"
                                     + "} else {\n"
                                     + "  alert(\"Your browser does not support Web Socket.\");\n"
                                     + "}\n"
                                     + "\n"
                                     + "function send(message) {\n"
                                     + "  if (!window.WebSocket) { return; }\n"
                                     + "  if (socket.readyState == WebSocket.OPEN) {\n"
                                     + "    socket.send(message);\n"
                                     + "    document.getElementById('msgForm').value = ''\n"
                                     + "  } else {\n"
                                     + "    alert(\"The socket is not open.\");\n"
                                     + "  }\n"
                                     + "}\n"
                                     + "</script>\n"
                                     + "<textarea id=\"responseText\" style=\"width:500px;height:300px;\"></textarea>\n"
                                     + "<form onsubmit=\"return false;\">\n"
                                     + "<input type=\"text\" name=\"message\" id=\"msgForm\"/><input type=\"button\" value=\"Send\"\n"
                                     + "       onclick=\"send(this.form.message.value)\" />\n"
                                     + "</form>\n"
                                     + "</body>\n"
                                     + "</html>\n", CharsetUtil.US_ASCII);
    }
}

改造后的页面效果长这样，虽然有些简陋，但还是可以收发消息。

最核心的就是WebSocketFrameHandler这个类了，所有的逻辑都是在这里面的，其实也不复杂，就是在连接建立后，给这个连接分配一个随机名字，将某个人发的消息转发到其他已有的连接上，另外及时清理掉断开的连接，防止资源泄露，代码很简单，相信你一看就懂。

复制

@Service
@Slf4j
@ChannelHandler.Sharable
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    // 我直接取了天龙八部里的名字，给每个聊天的人随机分配一个 
    private final List<String> names =
            List.of("刀白凤", "丁春秋", "马夫人", "马五德", "小翠", "于光豪", "巴天石", "不平道人", "邓百川", "风波恶",
                    "甘宝宝", "公冶乾", "木婉清", "少林老僧", "太皇太后", "天狼子", "天山童姥", "王语嫣", "乌老大",
                    "无崖子", "云岛主", "云中鹤", "止清", "白世镜", "包不同", "本参", "本观", "本相", "本因", "出尘子",
                    "冯阿三", "兰剑", "古笃诚", "过彦之", "平婆婆", "石清露", "石嫂", "司空玄", "司马林", "玄慈",
                    "玄寂", "玄苦", "玄难", "玄生", "玄痛", "叶二娘", "竹剑", "左子穆", "华赫艮", "乔峰", "李春来",
                    "李傀儡", "李秋水", "刘竹庄", "朴者和尚", "祁六三", "全冠清", "阮星竹", "西夏宫女", "许卓诚",
                    "朱丹臣", "努儿海", "阿碧", "阿洪", "阿胜", "阿朱", "阿紫", "波罗星", "陈孤雁", "鸠摩智", "来福儿",
                    "孟师叔", "宋长老", "苏星河", "苏辙", "完颜阿古打", "耶律洪基", "耶律莫哥", "耶律涅鲁古",
                    "耶律重元", "吴长风", "吴光胜", "吴领军", "辛双清", "严妈妈", "余婆婆", "岳老三", "张全祥",
                    "单伯山", "单季山", "单叔山", "单小山", "单正", "段延庆", "段誉", "段正淳", "段正明", "范禹",
                    "范百龄", "范骅", "苟读", "和里布", "何望海", "易大彪", "郁光标", "卓不凡", "宗赞王子", "哈大霸",
                    "姜师叔", "枯荣长老", "梦姑", "姚伯当", "神山上人", "神音", "狮鼻子", "室里", "项长老", "幽草",
                    "赵钱孙", "赵洵", "哲罗星", "钟灵", "钟万仇", "高升泰", "龚光杰", "贾老者", "康广陵", "秦红棉",
                    "虚竹", "容子矩", "桑土公", "唐光雄", "奚长老", "徐长老", "诸保昆", "崔百泉", "崔绿华", "符敏仪",
                    "黄眉和尚", "菊剑", "聋哑婆婆", "梅剑", "萧远山", "游骥", "游驹", "游坦之", "程青霜", "傅思归",
                    "葛光佩", "缘根", "智光大师", "鲍千灵", "褚万里", "瑞婆婆", "端木元", "黎夫人", "薛慕华", "慕容博",
                    "慕容复", "谭公", "谭婆", "谭青", "摘星子", "慧方", "慧观", "慧净", "慧真", "穆贵妃", "赫连铁树");

    // 名字到连接的映射
    private final Map<String, ChannelHandlerContext> name2ctx = new ConcurrentHashMap<>();

    // 连接到名字的映射
    private final Map<ChannelHandlerContext, String> ctx2name = new ConcurrentHashMap<>();

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {

        // 先分配当前没有在使用中的名字
        Optional<String> nameOp = names.stream().filter(x -> !name2ctx.containsKey(x)).findFirst();
        if (!nameOp.isPresent()) {
            // 如果没分配到名字，直接断开连接，这么写的话，同时在线的人数取决于名字列表的大小
            ctx.writeAndFlush(new TextWebSocketFrame("当前连接人数过多，请稍后重试！"));
            log.info("当前连接人数过多，请稍后重试！");
            ctx.close();
            return;
        }
        String name = nameOp.get();
        name2ctx.put(name, ctx);
        ctx2name.put(ctx, name);
        broadcast(name + "加入了群聊！");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        String name = ctx2name.getOrDefault(ctx, "");
        name2ctx.remove(name);
        ctx2name.remove(ctx);
        broadcast(name + "离开了群聊!");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        // 收到消息后将消息群发给所有人
        if (frame instanceof TextWebSocketFrame) {
            String request = ((TextWebSocketFrame) frame).text();
            String name = ctx2name.getOrDefault(ctx, "");
            broadcast(name + ":" + request);
        } else {
            String message = "unsupported frame type: " + frame.getClass().getName();
            throw new UnsupportedOperationException(message);
        }
    }

    /**
     * 将消息群发给所有在线的人
     */
    private void broadcast(String msg) {
        log.info("msg:{}", msg);
        name2ctx.entrySet().parallelStream().forEach(e -> {
            String name = e.getKey();
            ChannelHandlerContext ctx = e.getValue();
            if (ctx.channel().isActive()) {
                ctx.writeAndFlush(new TextWebSocketFrame(msg));
            } else {
                // 广播时清理掉不活跃的连接
                ctx2name.remove(ctx);
                name2ctx.remove(name);
            }
        });
    }
}

这里特别提醒下，想实现群聊，那WebSocketFrameHandler必须标记为Sharable，并且全局共享一个对象，所以需要注意下线程安全的问题，这里我都用了ConcurrentHashMap。
以上就是完整的代码了，有兴趣可以自己跑一跑，另外这个网络聊天室我已经部署的我的服务器上了，也可以直接点开体验下 http://xindoo.xyz:8083/。

赏

Recommend

彩虹女神跃长空,Go语言进阶之Go语言高性能Web框架Iris项目实战-登录与图形验证码(capt...

本田在美国EV市场再掀攻势

The Discovery and Beauty of Numbers in 4 Dimensions

能否重拾增长?2023冷年空调市场的趋势判断

马斯克减重18斤世界首富分享减肥秘籍 - CNMO

将中央办公室重新架构为数据中心

Will keeping MBA M2 attached to 10W or 30W charger while using reduce battery cy...

Sweet spot for MacBook Pro 14" configuration

一文看懂品牌反向营销的底层逻辑！

Sony Planning to Get Into Mobile Games With New Acquisition

About Joyk