Netty WebSocket客户端

参考：https://www.jianshu.com/p/f8f99f20d7f4

WebSocketClient.java

package com.flash.client;import com.flash.handler.WebSocketClientHandler;import com.flash.log.Log;import io.netty.bootstrap.Bootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.http.DefaultHttpHeaders;import io.netty.handler.codec.http.HttpHeaders;import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;import io.netty.handler.codec.http.websocketx.WebSocketVersion;import io.netty.util.concurrent.Future;import io.netty.util.concurrent.GenericFutureListener;import java.net.URI;public class WebSocketClient {public  boolean ConnWebSocketServerAsync(String svrUrl){Thread thd = new Thread(){@Overridepublic void run() {ConnWebSocketServer(svrUrl);}};thd.start();retu  true;}public  boolean ConnWebSocketServer(String svrUrl){EventLoopGroup client = new NioEventLoopGroup();try{Bootstrap bootstrap = new Bootstrap();bootstrap.group(client);bootstrap.option(ChannelOption.SO_KEEPALIVE, true);bootstrap.option(ChannelOption.TCP_NODELAY, true);bootstrap.channel(NioSocketChannel.class);bootstrap.handler(new WebSocketClientInit());URI wsURI = new URI(svrUrl);// new URI("ws://localhost:8899/ws");ChannelFuture cf = bootstrap.connect(wsURI.getHost(), wsURI.getPort()).sync();cf.addListener(new GenericFutureListener<ChannelFuture>() {@Overridepublic void operationComplete(ChannelFuture channelFuture) throws Exception {String log = "";log = String.format("连接websocket服务器: %s isSuccess=%s", svrUrl, channelFuture.isSuccess());System.out.println(log);if(channelFuture.isSuccess()){//进行握手Channel channel = channelFuture.channel();HttpHeaders httpHeaders = new DefaultHttpHeaders();WebSocketClientHandler handler = (WebSocketClientHandler)channel.pipeline().get("WebSocketClientHandler");WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(wsURI,WebSocketVersion.V13, (String)null, true,httpHeaders);handler.setHandshaker(handshaker);handshaker.handshake(channel);// 阻塞等待是否握手成功?//handler.handshakeFuture().sync();handler.handshakeFuture();}}});cf.channel().closeFuture().sync();retu  true;}catch (Exception ex){Log.ErrorLog(ex);}finally {client.shutdownGracefully();}retu  false;}// ConnWebSocketServer end}

WebSocketClientInit.java

package com.flash.client;import com.flash.handler.WebSocketClientHandler;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.socket.SocketChannel;import io.netty.handler.codec.http.HttpClientCodec;import io.netty.handler.codec.http.HttpObjectAggregator;public class WebSocketClientInit extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel channel) throws Exception {ChannelPipeline pipeline = channel.pipeline();pipeline.addLast("HttpClientCodec", new HttpClientCodec());pipeline.addLast("HttpObjectAggregator", new HttpObjectAggregator(1024*10));pipeline.addLast("WebSocketClientHandler", new WebSocketClientHandler());}}

WebSocketClientHandler.java

package com.flash.handler;import com.flash.log.Log;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelPromise;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.codec.http.FullHttpResponse;import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {private WebSocketClientHandshaker handshaker = null;private ChannelPromise handshakeFuture = null;public void handlerAdded(ChannelHandlerContext ctx) {this.handshakeFuture = ctx.newPromise();}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {Log.DebugLog("WebSocketClientHandler::channelRead0: ");// 握手协议返回，设置结束握手if (!this.handshaker.isHandshakeComplete()){FullHttpResponse response = (FullHttpResponse)msg;this.handshaker.finishHandshake(ctx.channel(), response);this.handshakeFuture.setSuccess();Log.DebugLog("WebSocketClientHandler::channelRead0 HandshakeComplete...");retu;}if (msg instanceof TextWebSocketFrame) {TextWebSocketFrame textFrame = (TextWebSocketFrame)msg;Log.DebugLog("WebSocketClientHandler::channelRead0 textFrame: " + textFrame.text());}if (msg instanceof CloseWebSocketFrame){Log.DebugLog("WebSocketClientHandler::channelRead0 CloseWebSocketFrame");}}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {//System.out.println("WebSocketClientHandler::channelInactive 服务端连接成功");}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("WebSocketClientHandler::exceptionCaught");cause.printStackTrace();ctx.channel().close();}public void setHandshaker(WebSocketClientHandshaker handshaker) {this.handshaker = handshaker;}public ChannelFuture handshakeFuture() {retu this.handshakeFuture;}public ChannelPromise getHandshakeFuture() {retu handshakeFuture;}public void setHandshakeFuture(ChannelPromise handshakeFuture) {this.handshakeFuture = handshakeFuture;}}