netty 实现websocket

1先加入jar
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>5.0.0.Alpha2</version>
</dependency>
2编写WebSocketServerHandler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.Constant;

import java.io.IOException;
import java.util.*;

public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
private WebSocketServerHandshaker handshaker;
private Map<String, Object> map = new HashMap<String, Object>();

/**
 * 接收客户端发送的消息
 *
 * @param channelHandlerContext ChannelHandlerContext
 * @param receiveMessage        消息
 */
@Override
protected void messageReceived(ChannelHandlerContext channelHandlerContext, Object receiveMessage) throws Exception {
    // 传统http接入 第一次需要使用http建立握手
    if (receiveMessage instanceof FullHttpRequest) {
        FullHttpRequest fullHttpRequest = (FullHttpRequest) receiveMessage;
        //  LOGGER.info("├ [握手]: {}", fullHttpRequest.uri());
        // 握手
        handlerHttpRequest(channelHandlerContext, fullHttpRequest);
        // 发送连接成功给客户端
        channelHandlerContext.channel().write(new TextWebSocketFrame("连接成功------------------"));
    }
    // WebSocket接入
    else if (receiveMessage instanceof WebSocketFrame) {
        WebSocketFrame webSocketFrame = (WebSocketFrame) receiveMessage;
        handlerWebSocketFrame(channelHandlerContext, webSocketFrame);
    }
}

/**
 * 第一次握手
 *
 * @param channelHandlerContext channelHandlerContext
 * @param req                   请求
 */
private void handlerHttpRequest(ChannelHandlerContext channelHandlerContext, FullHttpRequest req) {
    // 构造握手响应返回，本机测试
    WebSocketServerHandshakerFactory wsFactory
            = new WebSocketServerHandshakerFactory("ws://localhost:8080", null, false);
    // region 从连接路径中截取连接用户名
    String uri = req.uri();
    int i = uri.lastIndexOf("/");
    String userName = uri.substring(i + 1, uri.length());
    // endregion
    Channel connectChannel = channelHandlerContext.channel();
    // 加入在线用户
    map.put(userName, connectChannel);
    handshaker = wsFactory.newHandshaker(req);
    if (handshaker == null) {
        // 发送版本错误
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(connectChannel);
    } else {
        // 握手响应
        handshaker.handshake(connectChannel, req);
    }
}

/**
 * webSocket处理逻辑
 *
 * @param channelHandlerContext channelHandlerContext
 * @param frame                 webSocketFrame
 */
private void handlerWebSocketFrame(ChannelHandlerContext channelHandlerContext, WebSocketFrame frame) throws IOException {
    Channel channel = channelHandlerContext.channel();
    // region 判断是否是关闭链路的指令
    if (frame instanceof CloseWebSocketFrame) {
        // LOGGER.info("├ 关闭与客户端[{}]链接", channel.remoteAddress());
        handshaker.close(channel, (CloseWebSocketFrame) frame.retain());
        return;
    }
    // endregion
    // region 判断是否是ping消息
    if (frame instanceof PingWebSocketFrame) {
        // LOGGER.info("├ [Ping消息]");
        channel.write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }
    // endregion
    // region 纯文本消息
    if (frame instanceof TextWebSocketFrame) {
        String text = ((TextWebSocketFrame) frame).text();
        //  LOGGER.info("├ [{} 接收到客户端的消息]: {}", new Date(), text);
        channel.writeAndFlush(new TextWebSocketFrame("服务器返回消息：" + new Date()));
        System.out.println("消息已经返回");
    }
    // endregion
    // region 二进制消息 此处使用了MessagePack编解码方式

// if (frame instanceof BinaryWebSocketFrame) {
// BinaryWebSocketFrame binaryWebSocketFrame = (BinaryWebSocketFrame) frame;
// ByteBuf content = binaryWebSocketFrame.content();
// // LOGGER.info("├ [二进制数据]:{}", content);
// final int length = content.readableBytes();
// final byte[] array = new byte[length];
// content.getBytes(content.readerIndex(), array, 0, length);
// MessagePack messagePack = new MessagePack();
// WebSocketMessageEntity webSocketMessageEntity = messagePack.read(array, WebSocketMessageEntity.class);
// // LOGGER.info("├ [解码数据]: {}", webSocketMessageEntity);
// WebSocketUsers.sendMessageToUser(webSocketMessageEntity.getAcceptName(), webSocketMessageEntity.getContent());
// }
// endregion
}
}
3WebSocketServer 单独一个线程启动即可
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
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.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServer {
public void run(int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
//转化为http消息
pipeline.addLast("http-codec", new HttpServerCodec());
//多个分组组合一条http消息
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
//写入客户端
pipeline.addLast("http-chunked", new ChunkedWriteHandler());
pipeline.addLast("handler", new WebSocketServerHandler());
}
});
Channel ch = b.bind(port).sync().channel();
ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}

}

public static void main(String[] args) {
    int port = 8080;
    new WebSocketServer().run(port);
}

}
4html
<!DOCTYPE HTML>
<html>
<head>
<meta charset="utf-8">
<title>菜鸟教程(runoob.com)</title>

<script type="text/javascript">
    function WebSocketTest()
    {
        if ("WebSocket" in window)
        {
            alert("您的浏览器支持 WebSocket!");

            // 打开一个 web socket
            var ws = new WebSocket("ws://localhost:8080");

            ws.onopen = function()
            {
                // Web Socket 已连接上，使用 send() 方法发送数据
                ws.send("发送数据");
                alert("数据发送中...");
            };

            ws.onmessage = function (evt)
            {
                var received_msg = evt.data;
                console.log(received_msg);
                alert("数据已接收...");
            };

            ws.onclose = function()
            {
                // 关闭 websocket
                alert("连接已关闭...");
            };
        }

        else
        {
            // 浏览器不支持 WebSocket
            alert("您的浏览器不支持 WebSocket!");
        }
    }
</script>

</head>
<body>

<div id="sse">
<a href="javascript:WebSocketTest()">运行 WebSocket</a>
</div>

</body>
</html>
运行结果截图

image.png