使用netty框架实现websocket服务器端

刚学习如何使用netty，这是netty框架中的一个示例，你可以在netty的解压后文件夹路径netty-3.5.9.Final\src\main\java\org\jboss\netty\example\http\websocketx\server下找到所有的java文件
服务器端的websocket实现入口
WebSocketServer.java
package org.jboss.netty.example.http.websocketx.server;

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

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

//websocket到服务器端的链接，建议将localhost改为主机ip，各浏览器的支持情况
/**
* A HTTP server which serves Web Socket requests at:
*
* http://localhost:8080/websocket
*
* Open your browser at http://localhost:8080/, then the demo page will be loaded and a Web Socket connection will be
* made automatically.
*
* This server illustrates support for the different web socket specification versions and will work with:
*
* <ul>
* <li>Safari 5+ (draft-ietf-hybi-thewebsocketprotocol-00)
* <li>Chrome 6-13 (draft-ietf-hybi-thewebsocketprotocol-00)
* <li>Chrome 14+ (draft-ietf-hybi-thewebsocketprotocol-10)
* <li>Chrome 16+ (RFC 6455 aka draft-ietf-hybi-thewebsocketprotocol-17)
* <li>Firefox 7+ (draft-ietf-hybi-thewebsocketprotocol-10)
* <li>Firefox 11+ (RFC 6455 aka draft-ietf-hybi-thewebsocketprotocol-17)
* </ul>
*/
public class WebSocketServer {

    private final int port;//定义websocket端口

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

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

        // Set up the event pipeline factory.
        // 设置 Socket pipeline factory 
        bootstrap.setPipelineFactory(new WebSocketServerPipelineFactory());

        // Bind and start to accept incoming connections.
        // 启动服务，开始监听端口 
        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 + '/');
    }

    //main函数
    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
package org.jboss.netty.example.http.websocketx.server;

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 {
        // Create a default pipeline implementation.
        //创建默认的管道实现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
package org.jboss.netty.example.http.websocketx.server;

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;

/**
* Handles handshakes and messages
*/
public class WebSocketServerHandler extends SimpleChannelUpstreamHandler {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(WebSocketServerHandler.class);

    private static final String WEBSOCKET_PATH = "/websocket";

    private WebSocketServerHandshaker handshaker;

    //接收前台传来的message
    @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);
        }
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest req) throws Exception {
        // Allow only GET methods.    只允许GET方法
        if (req.getMethod() != GET) {
            sendHttpResponse(ctx, req, new DefaultHttpResponse(HTTP_1_1, FORBIDDEN));
            return;
        }

        // Send the demo page and favicon.ico
        if (req.getUri().equals("/")) {  //设置response
            HttpResponse res = new DefaultHttpResponse(HTTP_1_1, OK);

            ChannelBuffer content = WebSocketServerIndexPage.getContent(getWebSocketLocation(req));

            res.setHeader(CONTENT_TYPE, "text/html; charset=UTF-8");
            setContentLength(res, content.readableBytes());

            res.setContent(content);
            sendHttpResponse(ctx, req, res);
            return;
        } else if (req.getUri().equals("/favicon.ico")) {
            HttpResponse res = new DefaultHttpResponse(HTTP_1_1, NOT_FOUND);
            sendHttpResponse(ctx, req, res);
            return;
        }

        // Handshake  socket握手
        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) {

        // Check for closing 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()));
        }

        // Send the uppercase string back.
        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) {
        // Generate an error page if response status code is not OK (200).
        if (res.getStatus().getCode() != 200) {
            res.setContent(ChannelBuffers.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8));
            setContentLength(res, res.getContent().readableBytes());
        }

        // Send the response and close the connection if necessary.
        ChannelFuture f = ctx.getChannel().write(res);
        if (!isKeepAlive(req) || res.getStatus().getCode() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
        e.getCause().printStackTrace();
        e.getChannel().close();
    }

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

前台页面
index.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> 

<form onsubmit="return false;"> 
<h3>Receive :</h3>
<textarea id="responseText" style="width:500px;height:300px;"></textarea>
<br>
<h3>Send :</h3> <input type="text" name="message" value="Hello World!"/><input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" /> 

</form> 
</body> 
</html> 

注意，前台页面建立websocket时socket = new WebSocket("ws://localhost:8080/websocket"); 中的端口必须和服务器端开启服务占用的端口相同new WebSocketServer(port).run();
要import所有的类

主要应用到的netty类的简要说明，具体使用方法请参考API
Bootstrap  一个辅助类的初始化频道。这个类提供了常见的数据结构，它的子类初始化通道和他们的孩子通道使用常用的数据结构。
ServerBootstrap  创建一个新的服务器端的通道和接受传入连接的助手类。
ChannelPipeline  ChannelHandlers处理或拦截ChannelEvents频道的列表。 ChannelPipeline实现Intercepting Filter模式，给予用户完全控制事件如何处理，以及如何在管道中的ChannelHandlers互相交流的一种高级形式。
WebSocketServerHandshakerFactory  为服务器提供的适当的握手类

netty官网：https://netty.io/

评论

1 楼 parisspring 2013-09-27  

请问，websocketServer是怎么启动起来的，感觉一运行websocket的java应用程序后，前台的web页面就可以运行了，感觉很晕，前台的web页面应该是有web容器启动才能运行的。
另外一个问题是，服务器端的程序在项目1种，客户端index.html程序在项目2种，客户端的程序部署可以在其他的服务器上吧。我将websocketserver中的端口设置为8089就开始有报错，很奇怪