比尔云BierYun--阿里云最新优惠活动
阿里云优惠码丨阿里云代金券

Netty In Action中文版 – 第十一章:WebSocket

Netty In Action中文版 – 第十一章:WebSockethttp://www.bieryun.com/2182.html

本章介绍

  • WebSocket
  • ChannelHandler,Decoder and Encoder
  • 引导一个Netty基础程序
  • 测试WebSocket

“real-time-web”实时web现在随处可见,很多的用户希望能从web站点实时获取信息。Netty支持WebSocket实现,并包含了不同的版本,我们可以非常容易的实现WebSocket应用。使用Netty附带的WebSocket,我们不需要关注协议内部实现,只需要使用Netty提供的一些简单的方法就可以实现。本章将通过的例子应用帮助你来使用WebSocket并了解它是如何工作。

11.1 WebSockets some background

        关于WebSocket的一些概念和背景,可以查询网上相关介绍。这里不赘述。

11.2 面临的挑战

        要显示“real-time”支持的WebSocket,应用程序将显示如何使用Netty中的WebSocket实现一个在浏览器中进行聊天的IRC应用程序。你可能知道从Facebook可以发送文本消息到另一个人,在这里,我们将进一步了解其实现。在这个应用程序中,不同的用户可以同时交谈,非常像IRC(Internet Relay Chat,互联网中继聊天)。
上图显示的逻辑很简单:
  1. 一个客户端发送一条消息
  2. 消息被广播到其他已连接的客户端

它的工作原理就像聊天室一样,在这里例子中,我们将编写服务器,然后使用浏览器作为客户端。带着这样的思路,我们将会很简单的实现它。

11.3 实现

        WebSocket使用HTTP升级机制从一个普通的HTTP连接WebSocket,因为这个应用程序使用WebSocket总是开始于HTTP(s),然后再升级。什么时候升级取决于应用程序本身。直接执行升级作为第一个操作一般是使用特定的url请求。
        在这里,如果url的结尾以/ws结束,我们将只会升级到WebSocket,否则服务器将发送一个网页给客户端。升级后的连接将通过WebSocket传输所有数据。逻辑图如下:

11.3.1 处理http请求

        服务器将作为一种混合式以允许同时处理http和websocket,所以服务器还需要html页面,html用来充当客户端角色,连接服务器并交互消息。因此,如果客户端不发送/ws的uri,我们需要写一个ChannelInboundHandler用来处理FullHttpRequest。看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelFutureListener;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.DefaultFileRegion;
  6. import io.netty.channel.SimpleChannelInboundHandler;
  7. import io.netty.handler.codec.http.DefaultFullHttpResponse;
  8. import io.netty.handler.codec.http.DefaultHttpResponse;
  9. import io.netty.handler.codec.http.FullHttpRequest;
  10. import io.netty.handler.codec.http.FullHttpResponse;
  11. import io.netty.handler.codec.http.HttpHeaders;
  12. import io.netty.handler.codec.http.HttpResponse;
  13. import io.netty.handler.codec.http.HttpResponseStatus;
  14. import io.netty.handler.codec.http.HttpVersion;
  15. import io.netty.handler.codec.http.LastHttpContent;
  16. import io.netty.handler.ssl.SslHandler;
  17. import io.netty.handler.stream.ChunkedNioFile;
  18. import java.io.RandomAccessFile;
  19. /**
  20.  * WebSocket,处理http请求
  21.  * 
  22.  * @author c.k
  23.  * 
  24.  */
  25. public class HttpRequestHandler extends
  26.         SimpleChannelInboundHandler<FullHttpRequest> {
  27.     //websocket标识
  28.     private final String wsUri;
  29.     public HttpRequestHandler(String wsUri) {
  30.         this.wsUri = wsUri;
  31.     }
  32.     @Override
  33.     protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg)
  34.             throws Exception {
  35.         //如果是websocket请求,请求地址uri等于wsuri
  36.         if (wsUri.equalsIgnoreCase(msg.getUri())) {
  37.             //将消息转发到下一个ChannelHandler
  38.             ctx.fireChannelRead(msg.retain());
  39.         } else {//如果不是websocket请求
  40.             if (HttpHeaders.is100ContinueExpected(msg)) {
  41.                 //如果HTTP请求头部包含Expect: 100-continue,
  42.                 //则响应请求
  43.                 FullHttpResponse response = new DefaultFullHttpResponse(
  44.                         HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
  45.                 ctx.writeAndFlush(response);
  46.             }
  47.             //获取index.html的内容响应给客户端
  48.             RandomAccessFile file = new RandomAccessFile(
  49.                     System.getProperty(“user.dir”) + “/index.html”“r”);
  50.             HttpResponse response = new DefaultHttpResponse(
  51.                     msg.getProtocolVersion(), HttpResponseStatus.OK);
  52.             response.headers().set(HttpHeaders.Names.CONTENT_TYPE,
  53.                     “text/html; charset=UTF-8”);
  54.             boolean keepAlive = HttpHeaders.isKeepAlive(msg);
  55.             //如果http请求保持活跃,设置http请求头部信息
  56.             //并响应请求
  57.             if (keepAlive) {
  58.                 response.headers().set(HttpHeaders.Names.CONTENT_LENGTH,
  59.                         file.length());
  60.                 response.headers().set(HttpHeaders.Names.CONNECTION,
  61.                         HttpHeaders.Values.KEEP_ALIVE);
  62.             }
  63.             ctx.write(response);
  64.             //如果不是https请求,将index.html内容写入通道
  65.             if (ctx.pipeline().get(SslHandler.class) == null) {
  66.                 ctx.write(new DefaultFileRegion(file.getChannel(), 0, file
  67.                         .length()));
  68.             } else {
  69.                 ctx.write(new ChunkedNioFile(file.getChannel()));
  70.             }
  71.             //标识响应内容结束并刷新通道
  72.             ChannelFuture future = ctx
  73.                     .writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
  74.             if (!keepAlive) {
  75.                 //如果http请求不活跃,关闭http连接
  76.                 future.addListener(ChannelFutureListener.CLOSE);
  77.             }
  78.             file.close();
  79.         }
  80.     }
  81.     @Override
  82.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
  83.             throws Exception {
  84.         cause.printStackTrace();
  85.         ctx.close();
  86.     }
  87. }

11.3.2 处理WebSocket框架

        WebSocket支持6种不同框架,如下图:

我们的程序只需要使用下面4个框架:

  • CloseWebSocketFrame
  • PingWebSocketFrame
  • PongWebSocketFrame
  • TextWebSocketFrame

我们只需要显示处理TextWebSocketFrame,其他的会自动由WebSocketServerProtocolHandler处理,看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.channel.SimpleChannelInboundHandler;
  4. import io.netty.channel.group.ChannelGroup;
  5. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  6. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  7. /**
  8.  * WebSocket,处理消息
  9.  * @author c.k
  10.  *
  11.  */
  12. public class TextWebSocketFrameHandler extends
  13.         SimpleChannelInboundHandler<TextWebSocketFrame> {
  14.     private final ChannelGroup group;
  15.     public TextWebSocketFrameHandler(ChannelGroup group) {
  16.         this.group = group;
  17.     }
  18.     @Override
  19.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
  20.             throws Exception {
  21.         //如果WebSocket握手完成
  22.         if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
  23.             //删除ChannelPipeline中的HttpRequestHandler
  24.             ctx.pipeline().remove(HttpRequestHandler.class);
  25.             //写一个消息到ChannelGroup
  26.             group.writeAndFlush(new TextWebSocketFrame(“Client “ + ctx.channel()
  27.                     + ” joined”));
  28.             //将Channel添加到ChannelGroup
  29.             group.add(ctx.channel());
  30.         }else {
  31.             super.userEventTriggered(ctx, evt);
  32.         }
  33.     }
  34.     @Override
  35.     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg)
  36.             throws Exception {
  37.         //将接收的消息通过ChannelGroup转发到所以已连接的客户端
  38.         group.writeAndFlush(msg.retain());
  39.     }
  40. }

11.3.3 初始化ChannelPipeline

        看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.channel.Channel;
  3. import io.netty.channel.ChannelInitializer;
  4. import io.netty.channel.ChannelPipeline;
  5. import io.netty.channel.group.ChannelGroup;
  6. import io.netty.handler.codec.http.HttpObjectAggregator;
  7. import io.netty.handler.codec.http.HttpServerCodec;
  8. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  9. import io.netty.handler.stream.ChunkedWriteHandler;
  10. /**
  11.  * WebSocket,初始化ChannelHandler
  12.  * @author c.k
  13.  *
  14.  */
  15. public class ChatServerInitializer extends ChannelInitializer<Channel> {
  16.     private final ChannelGroup group;
  17.     public ChatServerInitializer(ChannelGroup group){
  18.         this.group = group;
  19.     }
  20.     @Override
  21.     protected void initChannel(Channel ch) throws Exception {
  22.         ChannelPipeline pipeline = ch.pipeline();
  23.         //编解码http请求
  24.         pipeline.addLast(new HttpServerCodec());
  25.         //写文件内容
  26.         pipeline.addLast(new ChunkedWriteHandler());
  27.         //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
  28.         //保证接收的Http请求的完整性
  29.         pipeline.addLast(new HttpObjectAggregator(64 * 1024));
  30.         //处理FullHttpRequest
  31.         pipeline.addLast(new HttpRequestHandler(“/ws”));
  32.         //处理其他的WebSocketFrame
  33.         pipeline.addLast(new WebSocketServerProtocolHandler(“/ws”));
  34.         //处理TextWebSocketFrame
  35.         pipeline.addLast(new TextWebSocketFrameHandler(group));
  36.     }
  37. }

WebSocketServerProtcolHandler不仅处理Ping/Pong/CloseWebSocketFrame,还和它自己握手并帮助升级WebSocket。这是执行完成握手和成功修改ChannelPipeline,并且添加需要的编码器/解码器和删除不需要的ChannelHandler。

        看下图:

ChannelPipeline通过ChannelInitializer的initChannel(…)方法完成初始化,完成握手后就会更改事情。一旦这样做了,WebSocketServerProtocolHandler将取代HttpRequestDecoder、WebSocketFrameDecoder13和HttpResponseEncoder、WebSocketFrameEncoder13。另外也要删除所有不需要的ChannelHandler已获得最佳性能。这些都是HttpObjectAggregator和HttpRequestHandler。下图显示ChannelPipeline握手完成:

我们甚至没注意到它,因为它是在底层执行的。以非常灵活的方式动态更新ChannelPipeline让单独的任务在不同的ChannelHandler中实现。

11.4 结合在一起使用

        一如既往,我们要将它们结合在一起使用。使用Bootstrap引导服务器和设置正确的ChannelInitializer。看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.bootstrap.ServerBootstrap;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.EventLoopGroup;
  7. import io.netty.channel.group.ChannelGroup;
  8. import io.netty.channel.group.DefaultChannelGroup;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.nio.NioServerSocketChannel;
  11. import io.netty.util.concurrent.ImmediateEventExecutor;
  12. import java.net.InetSocketAddress;
  13. /**
  14.  * 访问地址:http://localhost:2048
  15.  * 
  16.  * @author c.k
  17.  * 
  18.  */
  19. public class ChatServer {
  20.     private final ChannelGroup group = new DefaultChannelGroup(
  21.             ImmediateEventExecutor.INSTANCE);
  22.     private final EventLoopGroup workerGroup = new NioEventLoopGroup();
  23.     private Channel channel;
  24.     public ChannelFuture start(InetSocketAddress address) {
  25.         ServerBootstrap b = new ServerBootstrap();
  26.         b.group(workerGroup).channel(NioServerSocketChannel.class)
  27.                 .childHandler(createInitializer(group));
  28.         ChannelFuture f = b.bind(address).syncUninterruptibly();
  29.         channel = f.channel();
  30.         return f;
  31.     }
  32.     public void destroy() {
  33.         if (channel != null)
  34.             channel.close();
  35.         group.close();
  36.         workerGroup.shutdownGracefully();
  37.     }
  38.     protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {
  39.         return new ChatServerInitializer(group);
  40.     }
  41.     public static void main(String[] args) {
  42.         final ChatServer server = new ChatServer();
  43.         ChannelFuture f = server.start(new InetSocketAddress(2048));
  44.         Runtime.getRuntime().addShutdownHook(new Thread() {
  45.             @Override
  46.             public void run() {
  47.                 server.destroy();
  48.             }
  49.         });
  50.         f.channel().closeFuture().syncUninterruptibly();
  51.     }
  52. }
另外,需要将index.html文件放在项目根目录,index.html内容如下:

[html] view plain copy

  1. <html>
  2. <head>
  3. <title>Web Socket Test</title>
  4. </head>
  5. <body>
  6. <script type=“text/javascript”>
  7. var socket;
  8. if (!window.WebSocket) {
  9.   window.WebSocket = window.MozWebSocket;
  10. }
  11. if (window.WebSocket) {
  12.   socket = new WebSocket(“ws://localhost:2048/ws”);
  13.   socket.onmessage = function(event) {
  14.     var ta = document.getElementById(‘responseText’);
  15.     ta.value = ta.value + ‘\n’ + event.data
  16.   };
  17.   socket.onopen = function(event) {
  18.     var ta = document.getElementById(‘responseText’);
  19.     ta.value = “Web Socket opened!”;
  20.   };
  21.   socket.onclose = function(event) {
  22.     var ta = document.getElementById(‘responseText’);
  23.     ta.value = ta.value + “Web Socket closed”;
  24.   };
  25. } else {
  26.   alert(“Your browser does not support Web Socket.”);
  27. }
  28. function send(message) {
  29.   if (!window.WebSocket) { return; }
  30.   if (socket.readyState == WebSocket.OPEN) {
  31.     socket.send(message);
  32.   } else {
  33.     alert(“The socket is not open.”);
  34.   }
  35. }
  36. </script>
  37.     <form onsubmit=“return false;”>
  38.         <input type=“text” name=“message” value=“Hello, World!”><input
  39.             type=“button” value=“Send Web Socket Data”
  40.             onclick=“send(this.form.message.value)”>
  41.         <h3>Output</h3>
  42.         <textarea id=“responseText” style=“width: 500px; height: 300px;”></textarea>
  43.     </form>
  44. </body>
  45. </html>

最后在浏览器中输入:http://localhost:2048,多开几个窗口就可以聊天了。

11.5 给WebSocket加密

        上面的应用程序虽然工作的很好,但是在网络上收发消息存在很大的安全隐患,所以有必要对消息进行加密。添加这样一个加密的功能一般比较复杂,需要对代码有较大的改动。但是使用Netty就可以很容易的添加这样的功能,只需要将SslHandler加入到ChannelPipeline中就可以了。实际上还需要添加SslContext,但这不在本例子范围内。
        首先我们创建一个用于添加加密Handler的handler初始化类,看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.channel.Channel;
  3. import io.netty.channel.group.ChannelGroup;
  4. import io.netty.handler.ssl.SslHandler;
  5. import javax.net.ssl.SSLContext;
  6. import javax.net.ssl.SSLEngine;
  7. public class SecureChatServerIntializer extends ChatServerInitializer {
  8.     private final SSLContext context;
  9.     public SecureChatServerIntializer(ChannelGroup group,SSLContext context) {
  10.         super(group);
  11.         this.context = context;
  12.     }
  13.     @Override
  14.     protected void initChannel(Channel ch) throws Exception {
  15.         super.initChannel(ch);
  16.         SSLEngine engine = context.createSSLEngine();
  17.         engine.setUseClientMode(false);
  18.         ch.pipeline().addFirst(new SslHandler(engine));
  19.     }
  20. }

最后我们创建一个用于引导配置的类,看下面代码:

[java] view plain copy

  1. package netty.in.action;
  2. import io.netty.channel.Channel;
  3. import io.netty.channel.ChannelFuture;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.group.ChannelGroup;
  6. import java.net.InetSocketAddress;
  7. import javax.net.ssl.SSLContext;
  8. /**
  9.  * 访问地址:https://localhost:4096
  10.  * 
  11.  * @author c.k
  12.  * 
  13.  */
  14. public class SecureChatServer extends ChatServer {
  15.     private final SSLContext context;
  16.     public SecureChatServer(SSLContext context) {
  17.         this.context = context;
  18.     }
  19.     @Override
  20.     protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {
  21.         return new SecureChatServerIntializer(group, context);
  22.     }
  23.     /**
  24.      * 获取SSLContext需要相关的keystore文件,这里没有 关于HTTPS可以查阅相关资料,这里只介绍在Netty中如何使用
  25.      * 
  26.      * @return
  27.      */
  28.     private static SSLContext getSslContext() {
  29.         return null;
  30.     }
  31.     public static void main(String[] args) {
  32.         SSLContext context = getSslContext();
  33.         final SecureChatServer server = new SecureChatServer(context);
  34.         ChannelFuture future = server.start(new InetSocketAddress(4096));
  35.         Runtime.getRuntime().addShutdownHook(new Thread() {
  36.             @Override
  37.             public void run() {
  38.                 server.destroy();
  39.             }
  40.         });
  41.         future.channel().closeFuture().syncUninterruptibly();
  42.     }
  43. }

11.6 Summary

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

强烈推荐

高性能SSD云服务器ECS抗攻击,高可用云数据库RDS