Netty + websocket聊天室

程序处理逻辑

图1.png

启用websocket

从标准的HTTP或者HTTPS协议切换到WebSocket时，将会使用一种称为升级握手的机制。因此，使用WebSocket的应用程序将始终以HTTP/S作为开始，然后再执行升级。这个升级动作发生的确切时刻特定于应用程序；它可能会发生在启动时，也可能会发生在请求了某个特定的URL之后

约定：

• 如果被请求的 URL 以/ws 结尾，那么我们将会把该协议升级为 WebSocket；

• 否则，服务器将使用基本的 HTTP/S

服务器逻辑:

图2.png

处理http请求以及websocket

图3.png

//扩展 SimpleChannel-InboundHandler 以处理FullHttpRequest 消息
public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final String wsUri;
    private static final 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) {
            throw new IllegalStateException("Unable to locate WebsocketChatClient.html", e);
        }
    }

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

    @Override
    public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {

        if (wsUri.equalsIgnoreCase(request.uri())) {//如果请求了 WebSocket协议升级，则增加引用 计数（调用retain()方法），并将它传递给下一个ChannelInboundHandler
            ctx.fireChannelRead(request.retain()); //
        } else {
            if (HttpUtil.is100ContinueExpected(request)) {//处理 100 Continue请求以符合 HTTP 1.1 规范
                send100Continue(ctx); //
            }

            RandomAccessFile file = new RandomAccessFile(INDEX, "r");//

            HttpResponse response = new DefaultHttpResponse(request.protocolVersion(), HttpResponseStatus.OK);
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");

            boolean keepAlive = HttpUtil.isKeepAlive(request);

            if (keepAlive) { //如果请求了keep-alive，则添加所需要的HTTP 头信息
                response.headers().set(HttpHeaderNames.CONTENT_LENGTH, file.length());
                response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            }
            ctx.write(response); //将 HttpResponse写到客户端

            if (ctx.pipeline().get(SslHandler.class) == null) { //将 index.html写到客户端
                ctx.write(new DefaultFileRegion(file.getChannel(), 0, file.length()));
            } else {
                ctx.write(new ChunkedNioFile(file.getChannel()));
            }
            ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);//写LastHttpContent并冲刷至客户端
            if (!keepAlive) {//如果没有请求keep-alive，则在写操作完成后关闭 Channel
                future.addListener(ChannelFutureListener.CLOSE); //
            }

            file.close();
        }
    }

    private static 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 {
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "异常");
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {


    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { // 
        Channel incoming = ctx.channel();
        for (Channel channel : channels) {
            if (channel != incoming) {
                channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]" + msg.text()));
            } else {
                channel.writeAndFlush(new TextWebSocketFrame("[you]" + msg.text()));
            }
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception { // 
        Channel incoming = ctx.channel();

        // 广播
        channels.writeAndFlush(new TextWebSocketFrame("[SERVER] - " + incoming.remoteAddress() + " 加入"));

        channels.add(incoming);
        System.out.println("Client:" + incoming.remoteAddress() + "加入");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { // 
        Channel incoming = ctx.channel();

        // 广播
        channels.writeAndFlush(new TextWebSocketFrame("[SERVER] - " + incoming.remoteAddress() + " 离开"));

        System.out.println("Client:" + incoming.remoteAddress() + "离开");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { // 
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "在线");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception { // 
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "掉线");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) // 
            throws Exception {
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "异常");
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }


}

public class WebSocketChatServerInitializer extends ChannelInitializer<SocketChannel> {


    @Override
    public void initChannel(SocketChannel ch) throws Exception {//（2）
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(64*1024));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpRequestHandler("/ws"));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new TextWebSocketFrameHandler());

    }
}

Netty文档

ChannelGroup

线程安全的Set，包含开放的Channel，并在其上提供各种批量操作。 使用ChannelGroup，您可以将Channels划分为有意义的组（例如，基于每个服务或每个状态）。一个封闭的Channels会自动从集合中删除，因此您不必担心它的生命周期。 添加频道。 一个Channel可以属于多个ChannelGroup。

将消息广播到多个频道

如果需要将消息广播到多个频道，则可以添加与收件人关联的频道并调用write（Object）：

 ChannelGroup recipients =
         new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 recipients.add(channelA);
 recipients.add(channelB);
 ..
 recipients.write(Unpooled.copiedBuffer(
         "Service will shut down for maintenance in 5 minutes.",
         CharsetUtil.UTF_8));

使用ChannelGroup简化关机过程

如果ServerChannels和非ServerChannels都存在于同一ChannelGroup中，则首先对ServerChannels执行此组上所有请求的I / O操作，然后对其他Channels执行。

一次关闭服务器时，此规则非常有用：

 ChannelGroup allChannels =
         new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

 public static void main(String[] args) throws Exception {
     ServerBootstrap b = new ServerBootstrap(..);
     ...
     b.childHandler(new MyHandler());

     // Start the server
     b.getPipeline().addLast("handler", new MyHandler());
     Channel serverChannel = b.bind(..).sync();
     allChannels.add(serverChannel);

     ... Wait until the shutdown signal reception ...

     // Close the serverChannel and then all accepted connections.
     allChannels.close().awaitUninterruptibly();
 }

 public class MyHandler extends ChannelInboundHandlerAdapter {
      @Override
     public void channelActive(ChannelHandlerContext ctx) {
         // closed on shutdown.
         allChannels.add(ctx.channel());
         super.channelActive(ctx);
     }
 }

参考

<< Netty实战 >>