Netty Websocket服务器端和javascript客户端编程

[Java] Netty Websocket Server Javascript Clie

WebSocket协议的出现无疑是 HTML5 中最令人兴奋的功能特性之一，它能够很好地替代Comet技术以及Flash的XmlSocket来实现基于HTTP协议的双向通信。目前主流的浏览器，如Chrome、Firefox、IE10、Opera10、Safari等都已经支持WebSocket。另外，在服务端也出现了一些不错的WebSocket项目，如Resin、Jetty7、pywebsocket等。不过，本文将介绍的是如何使用强大的Netty框架（Netty-3.5.7.Final）来实现WebSocket服务端。

Netty3框架的性能优势已无需多说，但更让开发者舒心的是，Netty3还为大家提供了非常丰富的协议实现，包括HTTP、Protobuf、WebSocket等，开发者们可以很轻松的实现自己的Socket Server。按照Netty3的常规思路，我们需要准备以下3个文件：

1、WebSocketServer.java

2、WebSocketServerHandler.java

3、WebSocketServerPipelineFactory.java

以上3个文件分别包含了主程序的逻辑、服务的处理逻辑以及Socket Pipeline的设置逻辑。Java代码实现如下：

WebSocketServer.java

[java]
view plaincopyprint?

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;

import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

public class WebSocketServer

{
private final
int port;

public WebSocketServer(int port) {

this.port = port;
}

public void run() {

// 设置 Socket channel factory
ServerBootstrap bootstrap = new ServerBootstrap(

new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));

// 设置 Socket pipeline factory
bootstrap.setPipelineFactory(new WebSocketServerPipelineFactory());

// 启动服务，开始监听
bootstrap.bind(new InetSocketAddress(port));

// 打印提示信息
System.out.println("Web socket server started at port " + port +
'.');
System.out.println("Open your browser and navigate to http://localhost:" + port +
'/');
}

public static
void main(String[] args) {
int port;
if (args.length >
0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new WebSocketServer(port).run();
}
}

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

public class WebSocketServer
{
private final int port;

public WebSocketServer(int port) {
this.port = port;
}

public void run() {
// 设置 Socket channel factory
ServerBootstrap bootstrap = new ServerBootstrap(
new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));

// 设置 Socket pipeline factory
bootstrap.setPipelineFactory(new WebSocketServerPipelineFactory());

// 启动服务，开始监听
bootstrap.bind(new InetSocketAddress(port));

// 打印提示信息
System.out.println("Web socket server started at port " + port + '.');
System.out.println("Open your browser and navigate to http://localhost:" + port + '/');
}

public static void main(String[] args) {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new WebSocketServer(port).run();
}
}

WebSocketServerPipelineFactory.java

[java]
view plaincopyprint?

import static org.jboss.netty.channel.Channels.*;

import org.jboss.netty.channel.ChannelPipeline;

import org.jboss.netty.channel.ChannelPipelineFactory;

import org.jboss.netty.handler.codec.http.HttpChunkAggregator;

import org.jboss.netty.handler.codec.http.HttpRequestDecoder;

import org.jboss.netty.handler.codec.http.HttpResponseEncoder;

public class WebSocketServerPipelineFactory
implements ChannelPipelineFactory {
public ChannelPipeline getPipeline()
throws Exception {
// pipeline 的配置与 逻辑
ChannelPipeline pipeline = pipeline();
pipeline.addLast("decoder",
new HttpRequestDecoder());
pipeline.addLast("aggregator",
new HttpChunkAggregator(65536));
pipeline.addLast("encoder",
new HttpResponseEncoder());
pipeline.addLast("handler",
new WebSocketServerHandler());
return pipeline;
}
}

import static org.jboss.netty.channel.Channels.*;

import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.handler.codec.http.HttpChunkAggregator;
import org.jboss.netty.handler.codec.http.HttpRequestDecoder;
import org.jboss.netty.handler.codec.http.HttpResponseEncoder;

public class WebSocketServerPipelineFactory implements ChannelPipelineFactory {
public ChannelPipeline getPipeline() throws Exception {
// pipeline 的配置与 逻辑
ChannelPipeline pipeline = pipeline();
pipeline.addLast("decoder", new HttpRequestDecoder());
pipeline.addLast("aggregator", new HttpChunkAggregator(65536));
pipeline.addLast("encoder", new HttpResponseEncoder());
pipeline.addLast("handler", new WebSocketServerHandler());
return pipeline;
}
}

WebSocketServerHandler.java

[java]
view plaincopyprint?

import static org.jboss.netty.handler.codec.http.HttpHeaders.*;

import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*;

import static org.jboss.netty.handler.codec.http.HttpMethod.*;

import static org.jboss.netty.handler.codec.http.HttpResponseStatus.*;

import static org.jboss.netty.handler.codec.http.HttpVersion.*;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelFuture;

import org.jboss.netty.channel.ChannelFutureListener;

import org.jboss.netty.channel.ChannelHandlerContext;

import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;

import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import org.jboss.netty.handler.codec.http.DefaultHttpResponse;

import org.jboss.netty.handler.codec.http.HttpHeaders;

import org.jboss.netty.handler.codec.http.HttpRequest;

import org.jboss.netty.handler.codec.http.HttpResponse;

import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;

import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;

import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;

import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;

import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;

import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;

import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;

import org.jboss.netty.util.CharsetUtil;

public class WebSocketServerHandler
extends SimpleChannelUpstreamHandler
{
private static
final InternalLogger logger = InternalLoggerFactory
.getInstance(WebSocketServerHandler.class);

private static
final String WEBSOCKET_PATH = "/websocket";

private WebSocketServerHandshaker handshaker;

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)

throws Exception {
// 处理接受消息
Object msg = e.getMessage();
if (msg instanceof HttpRequest) {

handleHttpRequest(ctx, (HttpRequest) msg);
} else if (msg
instanceof WebSocketFrame) {
handleWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)

throws Exception {
// 处理异常情况
e.getCause().printStackTrace();
e.getChannel().close();
}

private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest req)

throws Exception {
// 只接受 HTTP GET 请求
if (req.getMethod() != GET) {
sendHttpResponse(ctx, req, new DefaultHttpResponse(HTTP_1_1,

FORBIDDEN));
return;
}

// Websocket 握手开始
WebSocketServerHandshakerFactory wsFactory =
new WebSocketServerHandshakerFactory(
getWebSocketLocation(req), null,
false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {

wsFactory.sendUnsupportedWebSocketVersionResponse(ctx.getChannel());

} else {
handshaker.handshake(ctx.getChannel(), req).addListener(
WebSocketServerHandshaker.HANDSHAKE_LISTENER);
}
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,

WebSocketFrame frame) {
// Websocket 握手结束
if (frame instanceof CloseWebSocketFrame) {

handshaker.close(ctx.getChannel(), (CloseWebSocketFrame) frame);
return;
} else if (frame
instanceof PingWebSocketFrame) {
ctx.getChannel().write(new PongWebSocketFrame(frame.getBinaryData()));

return;
} else if (!(frame
instanceof TextWebSocketFrame)) {
throw new UnsupportedOperationException(String.format("%s frame types not supported",

frame.getClass().getName()));
}

// 处理接受到的数据（转成大写）并返回
String request = ((TextWebSocketFrame) frame).getText();
if (logger.isDebugEnabled()) {
logger.debug(String.format("Channel %s received %s", ctx.getChannel().getId(), request));

}
ctx.getChannel().write(new TextWebSocketFrame(request.toUpperCase()));

}

private static
void sendHttpResponse(ChannelHandlerContext ctx,
HttpRequest req, HttpResponse res) {
// 返回 HTTP 错误页面
if (res.getStatus().getCode() !=
200) {
res.setContent(ChannelBuffers.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8));

setContentLength(res, res.getContent().readableBytes());
}

// 发送返回信息并关闭连接
ChannelFuture f = ctx.getChannel().write(res);
if (!isKeepAlive(req) || res.getStatus().getCode() !=
200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}

private static String getWebSocketLocation(HttpRequest req) {

return "ws://" + req.getHeader(HttpHeaders.Names.HOST) + WEBSOCKET_PATH;

}
}

import static org.jboss.netty.handler.codec.http.HttpHeaders.*;
import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*;
import static org.jboss.netty.handler.codec.http.HttpMethod.*;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.*;
import static org.jboss.netty.handler.codec.http.HttpVersion.*;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.CharsetUtil;

public class WebSocketServerHandler extends SimpleChannelUpstreamHandler
{
private static final InternalLogger logger = InternalLoggerFactory
.getInstance(WebSocketServerHandler.class);

private static final String WEBSOCKET_PATH = "/websocket";

private WebSocketServerHandshaker handshaker;

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
// 处理接受消息
Object msg = e.getMessage();
if (msg instanceof HttpRequest) {
handleHttpRequest(ctx, (HttpRequest) msg);
} else if (msg instanceof WebSocketFrame) {
handleWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
// 处理异常情况
e.getCause().printStackTrace();
e.getChannel().close();
}

private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest req)
throws Exception {
// 只接受 HTTP GET 请求
if (req.getMethod() != GET) {
sendHttpResponse(ctx, req, new DefaultHttpResponse(HTTP_1_1,
FORBIDDEN));
return;
}

// Websocket 握手开始
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
getWebSocketLocation(req), null, false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
wsFactory.sendUnsupportedWebSocketVersionResponse(ctx.getChannel());
} else {
handshaker.handshake(ctx.getChannel(), req).addListener(
WebSocketServerHandshaker.HANDSHAKE_LISTENER);
}
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,
WebSocketFrame frame) {
// Websocket 握手结束
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.getChannel(), (CloseWebSocketFrame) frame);
return;
} else if (frame instanceof PingWebSocketFrame) {
ctx.getChannel().write(new PongWebSocketFrame(frame.getBinaryData()));
return;
} else if (!(frame instanceof TextWebSocketFrame)) {
throw new UnsupportedOperationException(String.format("%s frame types not supported",
frame.getClass().getName()));
}

// 处理接受到的数据（转成大写）并返回
String request = ((TextWebSocketFrame) frame).getText();
if (logger.isDebugEnabled()) {
logger.debug(String.format("Channel %s received %s", ctx.getChannel().getId(), request));
}
ctx.getChannel().write(new TextWebSocketFrame(request.toUpperCase()));
}

private static void sendHttpResponse(ChannelHandlerContext ctx,
HttpRequest req, HttpResponse res) {
// 返回 HTTP 错误页面
if (res.getStatus().getCode() != 200) {
res.setContent(ChannelBuffers.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8));
setContentLength(res, res.getContent().readableBytes());
}

// 发送返回信息并关闭连接
ChannelFuture f = ctx.getChannel().write(res);
if (!isKeepAlive(req) || res.getStatus().getCode() != 200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}

private static String getWebSocketLocation(HttpRequest req) {
return "ws://" + req.getHeader(HttpHeaders.Names.HOST) + WEBSOCKET_PATH;
}
}

以上代码的逻辑还是比较清晰的：首先，在WebSocketServer中设置WebSocketServerPipelineFactory；然后，在WebSocketServerPipelineFactory中设置WebSocketServerHandler；接着，在WebSocketServerHandler处理请求并返回结果；其中，最重要的处理逻辑位于handleWebSocketFrame方法中，也就是把获取到的请求信息全部转化成大写并返回。最后，运行WebSocketServer.java就可以启动WebSocket服务，监听本地的8080端口。至此，WebSocket服务端已经全部准备就绪。这里，其实我们已经同时开发了一个简单的HTTP服务器，界面截图如下：



接下来，我们需要准备使用Javascript实现的WebSocket客户端，实现非常简单，Javascript代码实现如下：

websocket.html

[html]
view plaincopyprint?

<html><head><title>Web
Socket Client</title></head>

<body>

<script
type="text/javascript">

var socket;
if (!window.WebSocket) {
window.WebSocket =
window.MozWebSocket;
}
// Javascript Websocket Client
if (window.WebSocket) {
socket = new WebSocket("ws://localhost:8080/websocket");

socket.onmessage =
function(event) {
var ta = document.getElementById('responseText');

ta.value =
ta.value + '\n' + event.data
};
socket.onopen =
function(event) {
var ta = document.getElementById('responseText');

ta.value =
"Web Socket opened!";
};
socket.onclose =
function(event) {
var ta = document.getElementById('responseText');

ta.value =
ta.value + "Web Socket closed";
};
} else {
alert("Your browser does not support Web Socket.");
}
// Send Websocket data
function send(message) {
if (!window.WebSocket) { return; }
if (socket.readyState == WebSocket.OPEN) {

socket.send(message);
} else {
alert("The socket is not open.");
}
}
</script>

<h3>Send :</h3>

<form
onsubmit="return false;">

<input
type="text" name="message"
value="Hello World!"/><input
type="button"
value="Send Web Socket Data"
onclick="send(this.form.message.value)"
/>
<h3>Receive :</h3>

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

</form>

</body>

</html>

<html><head><title>Web Socket Client</title></head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
window.WebSocket = window.MozWebSocket;
}
// Javascript Websocket Client
if (window.WebSocket) {
socket = new WebSocket("ws://localhost:8080/websocket");
socket.onmessage = function(event) {
var ta = document.getElementById('responseText');
ta.value = ta.value + '\n' + event.data
};
socket.onopen = function(event) {
var ta = document.getElementById('responseText');
ta.value = "Web Socket opened!";
};
socket.onclose = function(event) {
var ta = document.getElementById('responseText');
ta.value = ta.value + "Web Socket closed";
};
} else {
alert("Your browser does not support Web Socket.");
}
// Send Websocket data
function send(message) {
if (!window.WebSocket) { return; }
if (socket.readyState == WebSocket.OPEN) {
socket.send(message);
} else {
alert("The socket is not open.");
}
}
</script>
<h3>Send :</h3>
<form onsubmit="return false;">
<input type="text" name="message" value="Hello World!"/><input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" />
<h3>Receive :</h3>
<textarea id="responseText" style="width:500px;height:300px;"></textarea>
</form>
</body>
</html>

以上Javascript代码的逻辑很好理解：即创建一个指向对应WebSocket地址（ws://localhost:8080/websocket）的Socket连接，进而进行发送和获取操作。其实，我们只需要把websocket.html文件放置到任意的HTTP服务器上，并打开对应URL地址，就可以看到以下的Demo界面：



输入文字“Hello World”并点击“Send Web Socket Data”按钮就可以向WebSocket的服务端发送消息了。从上图中我们还可以看到，在“Receive”下方的输出框中看到返回的消息（大写过的HELLO WORLD文字），这样一次基本的信息交互就完成了。当然，此时如果把服务端关闭，输出框中则会看到“Web Socket closed”信息。

来源：http://blog.csdn.net/shagoo/article/details/8028813