基于netty的websocket网络通信

使用websocket的作用，使前后端保持长连接通信，可互相发送消息。运行通过的代码如下：

一、后端：

1、引进jar包：

<properties>

<io.netty.netty-all.version>4.0.53.Final</io.netty.netty-all.version>

</properties>

<dependency>

<groupId>io.netty</groupId>

<artifactId>netty-all</artifactId>

<version>${io.netty.netty-all.version}</version>

</dependency>

2、新建类：

A、global类：

public class Global {

/**在Netty中提供了ChannelGroup接口，该接口继承Set接口，因此可以通过ChannelGroup可管理服务器端所有的连接的Channel，然后对所有的连接Channel广播消息。*/

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

}

B、ChildChannelHandler配置类：

public class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

@Override

protected void initChannel(SocketChannel e) throws Exception {

/**http解码,上述代码表示将请求和应答的消息编码或者解码为HTTP消息；*/

e.pipeline().addLast("http-codec",new HttpServerCodec());

/**上述代码表示将HTTP消息的多个部分组合成一条完整的HTTP消息；*/

e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));

/**上述代码表示向客户端发送HTML5文件，主要用于支持浏览器和服务端进行websocket通信；*/

e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());

/**MyWebSocketServerHandler 自己定义的类，用以处理传过来的数据*/

e.pipeline().addLast("handler",new MyWebSocketServerHandler());

}

}

C、WebSocketServer 启动类：

public class WebSocketServer {

public void run(){

EventLoopGroup bossGroup = new NioEventLoopGroup();

EventLoopGroup workGroup = new NioEventLoopGroup();

try {

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workGroup);

b.channel(NioServerSocketChannel.class);

b.childHandler(new ChildChannelHandler());

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

Channel ch = b.bind(8056).sync().channel();

ch.closeFuture().sync();

} catch (Exception e) {

e.printStackTrace();

}finally{

bossGroup.shutdownGracefully();

workGroup.shutdownGracefully();

}

}

}

D、具体业务逻辑类MyWebSocketServerHandler ：

public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object>{

private static final Logger logger = Logger

.getLogger(WebSocketServerHandshaker.class.getName());

private WebSocketServerHandshaker handshaker;

@Override

public void channelActive(ChannelHandlerContext ctx) throws Exception {

// 添加

Global.group.add(ctx.channel());

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

}

@Override

public void channelInactive(ChannelHandlerContext ctx) throws Exception {

// 移除

Global.group.remove(ctx.channel());

System.out.println("客户端与服务端连接关闭");

}

/*@Override

protected void messageReceived(ChannelHandlerContext ctx, Object msg)

throws Exception {

if (msg instanceof FullHttpRequest) {

handleHttpRequest(ctx, ((FullHttpRequest) msg));

} else if (msg instanceof WebSocketFrame) {

handlerWebSocketFrame(ctx, (WebSocketFrame) msg);

}

}*/

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

// TODO Auto-generated method stub

if (msg instanceof FullHttpRequest) {

System.out.println("进程：1");

handleHttpRequest(ctx, ((FullHttpRequest) msg));

} else if (msg instanceof WebSocketFrame) {

System.out.pri
d195
ntln("进程：2");

handlerWebSocketFrame(ctx, (WebSocketFrame) msg);

}

}

@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

ctx.flush();

}

private void handlerWebSocketFrame(ChannelHandlerContext ctx,

WebSocketFrame frame) {

System.out.println("进程：3");

// 判断是否关闭链路的指令

if (frame instanceof CloseWebSocketFrame) {

handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame

.retain());

}

// 判断是否ping消息

if (frame instanceof PingWebSocketFrame) {

ctx.channel().write(

new PongWebSocketFrame(frame.content().retain()));

return;

}

// 本例程仅支持文本消息，不支持二进制消息

if (!(frame instanceof TextWebSocketFrame)) {

System.out.println("本例程仅支持文本消息，不支持二进制消息");

throw new UnsupportedOperationException(String.format(

"%s frame types not supported", frame.getClass().getName()));

}

// 返回应答消息

String request = ((TextWebSocketFrame) frame).text();

System.out.println("服务端收到：" + request);

TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString()

+ "：" + request);

// 群发

Global.group.writeAndFlush(tws);

// 返回【谁发的发给谁】

// ctx.channel().writeAndFlush(tws);

}

private void handleHttpRequest(ChannelHandlerContext ctx,

FullHttpRequest req) {

System.out.println("进程：4");

if (!req.getDecoderResult().isSuccess()

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

sendHttpResponse(ctx, req, new DefaultFullHttpResponse(

HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));

return;

}

WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(

"ws://localhost:8056/websocket", null, false);

handshaker = wsFactory.newHandshaker(req);

if (handshaker == null) {

WebSocketServerHandshakerFactory

.sendUnsupportedWebSocketVersionResponse(ctx.channel());

} else {

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

}

}

private static void sendHttpResponse(ChannelHandlerContext ctx,

FullHttpRequest req, DefaultFullHttpResponse res) {

System.out.println("进程：5");

// 返回应答给客户端

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

ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),

CharsetUtil.UTF_8);

res.content().writeBytes(buf);

buf.release();

}

// 如果是非Keep-Alive，关闭连接

ChannelFuture f = ctx.channel().writeAndFlush(res);

if (!isKeepAlive(req) || res.getStatus().code() != 200) {

f.addListener(ChannelFutureListener.CLOSE);

}

}

private static boolean isKeepAlive(FullHttpRequest req) {

System.out.println("进程：8");

return false;

}

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

throws Exception {

cause.printStackTrace();

System.out.println("进程：7");

ctx.close();

}

@Override

protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {

// TODO Auto-generated method stub

System.out.println("进程：6");

}

}

E、启动接口websocket：

new WebSocketServer().run();

二、前端：

1.html 文件：

<!DOCTYPE html>

<html xmlns="http://www.w3.org/1999/xhtml">

<head>

<meta charset="utf-8">

<title>websocket</title>

<meta name="renderer" content="webkit">

<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">

<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">

<meta name="apple-mobile-web-app-status-bar-style" content="black">

<meta name="apple-mobile-web-app-capable" content="yes">

<meta name="format-detection" content="telephone=no">

<link rel="stylesheet" href="ag/css/layui.css" media="all"/>

<link rel="stylesheet" href="ag/css/bootstrap.min.css" media="all">

</head>

<body>

<form onSubmit="return false;">

<input type = "text" name="message" value="Netty The Sinper"/>

<br/><br/>

<input type="button" value="发送 WebSocket 请求消息" onClick="send(this.form.message.value)"/>

<hr color="blue"/>

<h3>服务端返回的应答消息</h3>

<textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>

</form>

</body>

<script type="text/javascript" src="ag/js/jquery.min.js"></script>

<script type="text/javascript" src="ag/js/bootstrap.min.js"></script>

<script type="text/javascript" src="ag/js/layui.js"></script>

<script type="text/javascript" src="ag/js/websocket.js"></script>

</html>

2、js文件：

var socket;

if(!window.WebSocket){

window.WebSocket = window.MozWebSocket;

}

if(window.WebSocket){

socket = new WebSocket("ws://localhost:8056/websocket");

socket.onmessage = function(event){

var ta = document.getElementById('responseText');

ta.value += event.data+"\r\n";

};

socket.onopen = function(event){

var ta = document.getElementById('responseText');

ta.value = "打开WebSoket 服务正常，浏览器支持WebSoket!"+"\r\n";

};

socket.onclose = function(event){

var ta = document.getElementById('responseText');

ta.value = "";

ta.value = "WebSocket 关闭"+"\r\n";

};

}else{

alert("您的浏览器不支持WebSocket协议！");

}

function send(message){

if(!window.WebSocket){return;}

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

socket.send(message);

}else{

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

}

}