Netty 系列八(基于 WebSocket 的简单聊天室).

一、前言

    之前写过一篇 Spring 集成 WebSocket 协议的文章 —— Spring消息之WebSocket ,所以对于 WebSocket 协议的介绍就不多说了,可以参考这篇文章。这里只做一些补充说明。另外,Netty 对 WebSocket 协议的支持要比 Spring 好太多了,用起来舒服的多。

    WebSocket 以帧的方式传输数据,每一帧代表消息的一部分。一个完整的消息可能会包含许多帧。

    由 IETF 发布的 WebSocket RFC,定义了 6 种帧, Netty 为它们每种都提供了一个 POJO 实现。下表列出了这些帧类型,并描述了它们的用法。

二、聊天室功能说明

    1、A、B、C 等所有用户都可以加入同一个聊天室。

    2、A 发送的消息,B、C 可以同时收到,但是 A 收不到自己发送的消息。

    3、当用户长时间没有发送消息,系统将把他踢出聊天室。

三、聊天室功能实现

  1、Netty 版本

<dependency>
     <groupId>io.netty</groupId>
     <artifactId>netty-all</artifactId>
     <version>5.0.0.Alpha2</version>
</dependency>

2、处理 HTTP 协议的 ChannelHandler —— 非 WebSocket 协议的请求,返回 index.html 页面

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final String wsUri;
    private static File INDEX;

    static {
        URL location = HttpRequestHandler.class.getProtectionDomain().getCodeSource().getLocation();
        try {
            String path = location.toURI() + "index.html";
            path = !path.contains("file:") ? path : path.substring(5);
            INDEX = new File(path);
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    public HttpRequestHandler(String wsUri) {
        this.wsUri = wsUri;
    }


    @Override
    protected void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        // 如果请求了Websocket,协议升级,增加引用计数(调用retain()),并将他传递给下一个 ChannelHandler
        // 之所以需要调用 retain() 方法,是因为调用 channelRead() 之后,资源会被 release() 方法释放掉,需要调用 retain() 保留资源
        if (wsUri.equalsIgnoreCase(request.uri())) {
            ctx.fireChannelRead(request.retain());
        } else {
            //处理 100 Continue 请求以符合 HTTP 1.1 规范
            if (HttpHeaderUtil.is100ContinueExpected(request)) {
                send100Continue(ctx);
            }
            // 读取 index.html
            RandomAccessFile randomAccessFile = new RandomAccessFile(INDEX, "r");
            HttpResponse response = new DefaultHttpResponse(request.protocolVersion(), HttpResponseStatus.OK);
            HttpHeaders headers = response.headers();
            //在该 HTTP 头信息被设置以后,HttpRequestHandler 将会写回一个 HttpResponse 给客户端
            headers.set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
            boolean keepAlive = HttpHeaderUtil.isKeepAlive(request);
            if (keepAlive) {
                headers.setLong(HttpHeaderNames.CONTENT_LENGTH, randomAccessFile.length());
                headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            }
            ctx.write(response);
            //将 index.html 写给客户端
            if (ctx.pipeline().get(SslHandler.class) == null) {
                ctx.write(new DefaultFileRegion(randomAccessFile.getChannel(), 0, randomAccessFile.length()));
            } else {
                ctx.write(new ChunkedNioFile(randomAccessFile.getChannel()));
            }
            //写 LastHttpContent 并冲刷至客户端,标记响应的结束
            ChannelFuture channelFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
            if (!keepAlive) {
                channelFuture.addListener(ChannelFutureListener.CLOSE);
            }
        }

    }

    private void send100Continue(ChannelHandlerContext ctx) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
        ctx.writeAndFlush(response);
    }

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

3、处理 WebSocket 协议的 ChannelHandler —— 处理 TextWebSocketFrame 的消息帧

/**
 * WebSocket 帧:WebSocket 以帧的方式传输数据,每一帧代表消息的一部分。一个完整的消息可能会包含许多帧
 */
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private final ChannelGroup group;

    public TextWebSocketFrameHandler(ChannelGroup group) {
        this.group = group;
    }

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //增加消息的引用计数(保留消息),并将他写到 ChannelGroup 中所有已经连接的客户端
        Channel channel = ctx.channel();
        //自己发送的消息不返回给自己
        group.remove(channel);
        group.writeAndFlush(msg.retain());
        group.add(channel);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //是否握手成功,升级为 Websocket 协议
        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            // 握手成功,移除 HttpRequestHandler,因此将不会接收到任何消息
            // 并把握手成功的 Channel 加入到 ChannelGroup 中
            ctx.pipeline().remove(HttpRequestHandler.class);
            group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined"));
            group.add(ctx.channel());
        } else if (evt instanceof IdleStateEvent) {
            IdleStateEvent stateEvent = (IdleStateEvent) evt;
            if (stateEvent.state() == IdleState.READER_IDLE) {
                group.remove(ctx.channel());
                ctx.writeAndFlush(new TextWebSocketFrame("由于您长时间不在线,系统已自动把你踢下线!")).addListener(ChannelFutureListener.CLOSE);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

WebSocket 协议升级完成之后, WebSocketServerProtocolHandler 将会把 HttpRequestDecoder 替换为 WebSocketFrameDecoder,把 HttpResponseEncoder 替换为WebSocketFrameEncoder。为了性能最大化,它将移除任何不再被 WebSocket 连接所需要的 ChannelHandler。这也包括了 HttpObjectAggregator 和 HttpRequestHandler 。

    4、ChatServerInitializer —— 多个 ChannelHandler 合并成 ChannelPipeline 链

public class ChatServerInitializer extends ChannelInitializer<Channel> {

    private final ChannelGroup group;
    private static final int READ_IDLE_TIME_OUT = 60; // 读超时
    private static final int WRITE_IDLE_TIME_OUT = 0;// 写超时
    private static final int ALL_IDLE_TIME_OUT = 0; // 所有超时

    public ChatServerInitializer(ChannelGroup group) {
        this.group = group;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
        // 处理那些不发送到 /ws URI的请求
        pipeline.addLast(new HttpRequestHandler("/ws"));
        // 如果被请求的端点是 "/ws",则处理该升级握手
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        // //当连接在60秒内没有接收到消息时,进会触发一个 IdleStateEvent 事件,被 HeartbeatHandler 的 userEventTriggered 方法处理
        pipeline.addLast(new IdleStateHandler(READ_IDLE_TIME_OUT, WRITE_IDLE_TIME_OUT, ALL_IDLE_TIME_OUT, TimeUnit.SECONDS));
        pipeline.addLast(new TextWebSocketFrameHandler(group));

    }
}

tips:上面这些开箱即用 ChannelHandler 的作用,我就不一一介绍了,可以参考上一篇文章

  5、引导类 ChatServer

public class ChatServer {

    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
    private final EventLoopGroup group = new NioEventLoopGroup();
    private Channel channel;

    public ChannelFuture start(InetSocketAddress address) {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChatServerInitializer(channelGroup));
        ChannelFuture channelFuture = bootstrap.bind(address);
        channelFuture.syncUninterruptibly();
        channel = channelFuture.channel();
        return channelFuture;
    }

    public void destroy() {
        if (channel != null) {
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public static void main(String[] args) {
        final ChatServer chatServer = new ChatServer();
        ChannelFuture channelFuture = chatServer.start(new InetSocketAddress(9999));
        // 返回与当前Java应用程序关联的运行时对象
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                chatServer.destroy();
            }
        });
        channelFuture.channel().closeFuture().syncUninterruptibly();
    }

}

三、效果展示

    在浏览器中输入 http://127.0.0.1:9999 即可看到预先准备好的 index.html 页面;访问 ws://127.0.0.1:9999/ws (可随意找一个 WebSocket 测试工具测试)即可加入聊天室。

有点 low 的聊天室总算是完成了,算是 Netty 对 HTTP 协议和 WebSocket 协议的一次实践吧!虽然功能欠缺,但千里之行,始于足下!不积硅步,无以至千里;不积小流,无以成江海!

参考资料:《Netty IN ACTION》

演示源代码:https://github.com/JMCuixy/NettyDemo/tree/master/src/main/java/org/netty/demo/chatroom

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Java编程技术

Netty学习笔记(一)

Netty是一种可以轻松快速的开发类似协议服务器和客户端网络应用程序的NIO客户端服务器框架,它大大简化了TCP或者UDP服务器的网络编程,但是你仍然可以访问和...

1562
来自专栏码匠的流水账

jvm排查工具箱jvm-tools

本文主要介绍的是一款jvm排查工具箱:jvm-tools。除了对基本jvm封装外,还提供了jmx访问以及火焰图的生成。

1091
来自专栏知识分享

4-MSP430定时器_定时器中断

一开始没写好就上传了,,,,,,,,这次来个全的 自己学MSP430是为了写一篇关于PID的文章,需要430在proteus上做仿真,一则认为在自动控制算法上P...

3496
来自专栏智能大石头

线程池ThreadPool及Task调度机制分析

近1年,偶尔发生应用系统启动时某些操作超时的问题,特别在使用4核心Surface以后。笔记本和台式机比较少遇到,服务器则基本上没有遇到过。

1120
来自专栏Android 研究

APK安装流程详解2——PackageManager简介

俗话说的好,得中原者,得天下,那么想要了解Android的安装了流程就不得不提及一个重要的类"PackageManager"我们就先来了解这两个类

2833
来自专栏deepcc

linux中nodejs后台运行工具forever

3088
来自专栏小樱的经验随笔

自己手动复现一个熊猫烧香病毒

最近逛了一下 bilibili ,偶然的一次机会,我在 bilibili 上看到了某个 up 主分享了一个他自己仿照熊猫病毒的原型制作的一个病毒的演示视频,虽然...

5932
来自专栏逸鹏说道

我这么玩Web Api(一)

帮助页面或用户手册(Microsoft and Swashbuckle Help Page) 前言   你需要为客户编写Api调用手册?你需要测试你的Api接口...

3085
来自专栏郭霖

巧用Android网络通信技术,在网络上直接传输对象

要做一个优秀的Android应用,使用到网络通信技术是必不可少的,很难想象一款没有网络交互的软件最终能发展得多成功。那么我们来看一下,一般Android应用程序...

2256
来自专栏Android 研究

APK安装流程详解12——PMS中的新安装流程上(拷贝)

从上面一片文章我们知道InstallAppProgress里面最后更新的代码是调用到PackageManager#installPackageWithVerif...

1351

扫码关注云+社区

领取腾讯云代金券