功能介绍Netty开发服务器

HTML实现客户端

实现服务端与客户端时实时交互

开发步骤

1.导包

io.netty

netty‐all

5.0.0.Alpha2

2.工程配置文件：NettyConfig

/**

* 这里放的是工程中相应的配置

*/

public class NettyConfig{

/**

* 用于存储每一个客户端接入进来时的channel对象

*/

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

}

3.MyWebSocketHandler

public class MyWebSocketHandler extends SimpleChannelInboundHandler { private WebSocketServerHandshaker handshaker;

//请求路径

private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket";

//工程出现异常的时候调用@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

cause.printStackTrace(); ctx.close(); //关闭当前连接

}

//客户端与服务器创建连接的时候调用@Override

public void channelActive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.add(ctx.channel());

System.out.println("客户端与服务器连接开启");

}

//客户端与服务器断开连接的时候调用@Override

public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.remove(ctx.channel());

System.out.println("客户端与服务器断开连接");

}

//服务端接收客户端发送过来的数据结束之后调用@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception{ ctx.flush(); //进行数据清除

}

//服务端处理客户端调用的核心方法

protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception {

//处理客户端向服务端发起http握手请求业务if(msg instanceof FullHttpRequest){

handHttpRequest(context,(FullHttpRequest) msg);

}else if(msg instanceof WebSocketFrame){

//处理websocket连接业务handWebSocketFrme(context,(WebSocketFrame)msg);

}

}

/**

* 处理客户端与服务端之前的websocket业务

* @param ctx

* @param frame

*/

private void handWebSocketFrme(ChannelHandlerContext ctx,WebSocketFrame frame){

//判断是否是关闭websocket的指令

if(frame instanceof CloseWebSocketFrame){ handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());

}

//判断是否是Ping消息

if(frame instanceof PingWebSocketFrame){

//返回一个胖信息

ctx.channel().write(newPongWebSocketFrame(frame.content().retain())); return;

}

//判断是否是二进制消息，如果是二进制消息，抛出异常if(!(frame instanceof TextWebSocketFrame)){

System.out.println("目前我们不支持二进制消息");

throw new RuntimeException(this.getClass().getName()+"不支持的消息");

}

//返回应答消息

//获取客户端向服务端发送的消息

String request = ((TextWebSocketFrame)frame).text(); System.out.println("服务端收到客户端的消息===>"+request); TextWebSocketFrame tws = new TextWebSocketFrame(new

Date().toString()+ctx.channel().id()+"===>"+request);

//群发，服务端向每个上来的客户端群发消息NettyConfig.group.writeAndFlush(tws);

}

/**

* 处理客户端向服务器发起http握手请求的业务

* @param ctx

* @param req

*/

private void handHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req){

// req.headers().get()

// req.decoderResult().isSuccess()

//不是websocket,就不是客户端发给服务器的Http请求if(!req.decoderResult().isSuccess()|| !

("websocket").equals(req.headers().get("Upgrade"))){

sendHttpResponse(ctx,req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));

return;

}

WebSocketServerHandshakerFactory wsFactory =

newWebSocketServerHandshakerFactory(WEB_SOCKET_URL,null,false); handshaker = wsFactory.newHandshaker(req);

if(handshaker==null){

WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());

}else{

handshaker.handshake(ctx.channel(),req);

}

}

/**

* 服务端向客户端响应消息

* @param ctx

* @param req

* @param res

*/

private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res){

if(res.status().code()!=200){

ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);

res.content().writeBytes(buf); buf.release();

}

//服务器向客户端发送数据

ChannelFuture f = ctx.channel().writeAndFlush(res); if(res.status().code()!=200){

f.addListener(ChannelFutureListener.CLOSE);

}

}

}

4 初始化连接各个组件:MyWebSocketChannelHandler

/**

* 初始化连接各个组件

*/

public class MyWebSocketChannelHandlerextends ChannelInitializer{

protected void initChannel(SocketChannel e) throws Exception { e.pipeline().addLast("http‐codec",new HttpServerCodec()); e.pipeline().addLast("aggregator",new HttpObjectAggregator(66636)); e.pipeline().addLast("http‐chunked",new ChunkedWriteHandler()); e.pipeline().addLast("handler",new MyWebSocketHandler());

}

}

5 程序入口

/**

* 程序的入口，负责启动应用

*/

public class Main {

public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workGroup = new NioEventLoopGroup();try{

ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup,workGroup); b.channel(NioServerSocketChannel.class); b.childHandler(new MyWebSocketChannelHandler());

System.out.println("服务端开启等待客户端连接");

Channel ch = b.bind(8888).sync().channel(); ch.closeFuture().sync();

} catch (Exception e) {

e.printStackTrace();

} finally {

//优雅的退出程序bossGroup.shutdownGracefully(); workGroup.shutdownGracefully();

}

}

}

6 web端测试代码

WebSocket客户端

var socket; if(!window.WebSocket){

window.WebSocket = window.MozWebSocket;

}

if(window.WebSocket){

socket = new WebSocket("ws://localhost:8888/websocket"); socket.onmessage = function (event) {

var ta = document.getElementById("responseContext"); ta.value += event.data +"\r\n";

}

socket.onopen = function (event) {

var ta = document.getElementById("responseContext"); ta.value = "您浏览器支持webSoceket，请进行后续 操作\r\n";

}

socket.onclose = function (event) {

var ta = document.getElementById("responseContext"); ta.value = "";

ta.value = "webSoceket连接已经关闭";

}

}else{

console.debug("浏览器不支持webSocket")

}

function send() {

var message = document.getElementById("message").value; if(!window.WebSocket){

return;

}

if(socket.readyState == WebSocket.OPEN){ socket.send(message)

}else{

alert("连接没有建立成功！");

}

}

---

客户端接收到服务端返回的消息