Netty基于WebSocket实现的简单Web聊天室

服务端代码

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

Java代码如下:

WebSocketNettyServer 服务启动入口
WebSocketChannelInitializer 自定义处理通道
ChatHandler 消息处理

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class WebSocketNettyServer {

    public static void main(String[] args) {
        // 创建两个线程池
        NioEventLoopGroup mainGrp = new NioEventLoopGroup(); // 主线程池
        NioEventLoopGroup subGrp = new NioEventLoopGroup(); // 从线程池

        try {
            // 创建Netty服务器启动对象
            ServerBootstrap serverBootstrap = new ServerBootstrap();

            // 初始化服务器启动对象
            serverBootstrap
                    // 指定使用上面创建的两个线程池
                    .group(mainGrp, subGrp)
                    // 指定Netty通道类型
                    .channel(NioServerSocketChannel.class)
                    // 指定通道初始化器用来加载当Channel收到事件消息后，
                    // 如何进行业务处理
                    .childHandler(new WebSocketChannelInitializer());

            // 绑定服务器端口，以同步的方式启动服务器
            ChannelFuture future = serverBootstrap.bind(9090).sync();
            // 等待服务器关闭
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 优雅关闭服务器
            mainGrp.shutdownGracefully();
            subGrp.shutdownGracefully();
        }

    }
}

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
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.stream.ChunkedWriteHandler;

/**
 * 通道初始化器
 * 用来加载通道处理器（ChannelHandler)
 */
public class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {

    // 初始化通道
    // 在这个方法中去加载对应的ChannelHandler
    protected void initChannel(SocketChannel ch) {
        // 获取管道，将一个一个的ChannelHandler添加到管道中
        ChannelPipeline pipeline = ch.pipeline();

        // 添加一个http的编解码器
        pipeline.addLast(new HttpServerCodec());
        // 添加一个用于支持大数据流的支持
        pipeline.addLast(new ChunkedWriteHandler());
        // 添加一个聚合器，这个聚合器主要是将HttpMessage聚合成FullHttpRequest/Response
        pipeline.addLast(new HttpObjectAggregator(1024 * 64));

        // 需要指定接收请求的路由
        // 必须使用以ws后缀结尾的url才能访问
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));

        // 添加自定义的Handler
        pipeline.addLast(new ChatHandler());
    }
}

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.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;

public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    // 用来保存所有的客户端连接
    private static final ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm");

    // 当Channel中有新的事件消息会自动调用
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        // 当接收到数据后会自动调用

        // 获取客户端发送过来的文本消息
        String text = msg.text();
        System.out.println("接收到消息数据为：" + text);
        Channel channel = ctx.channel();
        InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
        String hostName = inetSocketAddress.getHostName();
        Integer port = inetSocketAddress.getPort();
        String username = String.format("%s:%s", hostName, port);
        String chatContent = String.format("[%s] [%s] [%s]", sdf.format(new Date()), username, text);
        for (Channel client : clients) {
            // 将消息发送到所有的客户端
            client.writeAndFlush(new TextWebSocketFrame(chatContent));
        }
    }

    // 当有新的客户端连接服务器之后，会自动调用这个方法
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 将新的通道加入到clients
        clients.add(ctx.channel());
    }
}

客户端代码

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>在线聊天室</title>
</head>
<body>
    <input type="text" id="message">
    <input type="button" value="发送消息" onclick="sendMsg()">

    接收到的消息:
    <p id="server_message" style="background-color: #AAAAAA"></p>

    <script>

        var websocket = null;

        // 判断当前浏览器是否支持websocket
        if(window.WebSocket) {
            websocket = new WebSocket("ws://127.0.0.1:9090/ws");

            websocket.onopen = function() {
                console.log("建立连接.");
            }
            websocket.onclose = function() {
                console.log("断开连接");
            }
            websocket.onmessage = function(e) {
                console.log("接收到服务器消息:" + e.data);
                var server_message = document.getElementById("server_message");
                server_message.innerHTML += e.data + "<br/>";
            }
        }
        else {
            alert("当前浏览器不支持web socket");
        }

        function sendMsg() {
            var message = document.getElementById("message");
            websocket.send(message.value);
        }
    </script>
</body>
</html>

体验一下

开启两个浏览器标签页作为两个客户端,并发起聊天

聊天室用户: (127.0.0.1:49790)

image

聊天室用户: (127.0.0.1:52341)

image

后端服务日志

Connected to the target VM, address: '127.0.0.1:58534', transport: 'socket'
接收到消息数据为：上线上线
接收到消息数据为：我等到花都谢了
接收到消息数据为：啥时候来的?
接收到消息数据为：别抬杠了,上分要紧