1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
package study.netty;

import java.util.TimerTask;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.Timer;

public class NettyWebSocketServer {
	public static void main(String[] args) {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		ServerBootstrap bootstrap = new ServerBootstrap();
		bootstrap.group(bossGroup,workerGroup)
				.channel(NioServerSocketChannel.class)
				.childHandler(new ChannelInitializer<SocketChannel>() {

					@Override
					protected void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast("http-codec",new HttpServerCodec())
									.addLast("aggregator",new HttpObjectAggregator(65535))
									.addLast("http-chunked",new ChunkedWriteHandler())
									.addLast("handler",new WebSocketServerHandler());
					}
				});
		try {
			bootstrap.bind(8080).sync().channel().closeFuture().sync();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally {
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}
	private static class WebSocketServerHandler extends SimpleChannelInboundHandler<Object>{
		private WebSocketServerHandshaker handshaker;
		@Override
		protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
			if (msg instanceof FullHttpRequest) {//建立连接的请求
				handleHttpRequest(ctx,(FullHttpRequest)msg);
			}else if (msg instanceof WebSocketFrame){//WebSocket
				handleWebsocketFrame(ctx,(WebSocketFrame)msg);
			}
		}
		private void handleWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
			if (frame instanceof CloseWebSocketFrame) {//关闭
				handshaker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
			}else if (frame instanceof PingWebSocketFrame) {//ping消息
				ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
			}else if (frame instanceof TextWebSocketFrame) {//文本消息
				String request = ((TextWebSocketFrame)frame).text();
				ctx.channel().write(new TextWebSocketFrame("websocket return:"+request));
			}
		}
		private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
			if(request.getDecoderResult().isSuccess()&&"websocket".equals(request.headers().get("Upgrade"))) {
				System.out.println("create WebSocket connection");
				WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory("ws://localhost:8080/websocket", null, false);
				handshaker = factory.newHandshaker(request);//通过创建请求生成一个握手对象
				if(handshaker != null) {
					handshaker.handshake(ctx.channel(),request);
				}
			}
		}
		@Override
		public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
			ctx.flush();
		}
		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
			cause.printStackTrace();
			ctx.close();
		}
		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			new java.util.Timer().schedule(new TimerTask() {
				
				@Override
				public void run() {
					if(handshaker !=null) {
						ctx.channel().write(new TextWebSocketFrame("server:主动给客户端发消息"));
						ctx.flush();
					}
				}
			}, 1000,1000);
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
<html>
	<head>
		<title>test</title>
	</head>
	<body>
		<form οnsubmit="return false;">
			<input type="text" name="message">
			<button οnclick="send(this.form.message.value)">send</button>
			<br>
			<h3>return messages</h3>
			<textarea id="returns" rows="10" cols="30"></textarea>
		</form>
	</body>
	<script type="text/javascript">
		var socket
		if(!window.WebSocket){
			window.WebSocket = window.MozWebSocket
		}
		if(window.WebSocket){
			socket = new WebSocket("ws://localhost:8080/websocket")
			socket.onmessage = function(event){
				var returns = document.getElementById("returns")
				returns.value = returns.value + "n"
				returns.value = returns.value + event.data
			}
			socket.onopen = function(event){
				var returns = document.getElementById("returns")
				returns.value = returns.value + "n"
				returns.value = returns.value + "已经连接上WebSocket服务器"
			}
			socket.onclose = function(event){
				var returns = document.getElementById("returns")
				returns.value = returns.value + "n"
				returns.value = returns.value + "断开与Socket服务器的连接"
			}
		}
		function send(msg){
			if(!socket){
				alert()
				return
			}
			if(socket.readyState == WebSocket.OPEN){
				socket.send(msg)
			}else{
				alert("WebSocket还未连接成功")
			}
		}
	</script>
</html>