Java封装netty websocket server

Java封装netty websocket server

netty是一个高效的NIO框架,用netty封装的websocket服务器不仅稳定性能也非常优秀,由于netty会用线程绑定连接(通道),在游戏开发中可以利用这种特性进行单用户无锁开发,可以大量的减少对锁的使用,提高游戏的吞吐量,这里对netty进行简单封装,方便使用

git地址

https://gitee.com/ichiva/netty-websocket-server.git

介绍

封装netty用于快速创建websocket服务器

快速入门

创建FastNettyWebSocket实例,传入消息处理器,并开启监听

    public static void main(String[] args) {
        new FastNettyWebSocketServer(new WebSocket() {
            @Override
            public void onMessage(WebSocketSession session, String message) {
                System.out.println("收到:" + message);
                send(session,"收到,over");
            }
        }).listener(8080);
    }

实现细节

定义WebSocketServer接口

public interface WebSocketServer {
    
    default void onOpen(WebSocketSession session){

    }

    
    void onMessage(WebSocketSession session, String message);

    
    default void onMessage(WebSocketSession session, byte[] message) {

    }

    
    default void send(WebSocketSession session, String message){
        session.getChannel().writeAndFlush(
                new TextWebSocketframe(message)
        );
    }

    
    default void send(WebSocketSession session, byte[] message){
        session.getChannel().writeAndFlush(new BinaryWebSocketframe(
                Unpooled.buffer().writeBytes(message)
        ));
    }

    
    default void onClose(WebSocketSession session) {

    }

    
    default void onError(WebSocketSession session, Throwable e){

    }
}

定义配置文件

public interface NettyWebsocketServerConfig {

    
    default  NioEventLoopGroup getWorkerGroup(){
        return new NioEventLoopGroup();
    }

    
    default NioEventLoopGroup getBoosGroup(){
        return new NioEventLoopGroup(1);
    }

    ChannelHandler getChildHandler();

    
    default int getPort(){ return 8080; }

}

WebSocketServer (核心)实现,用于启动netty和关闭netty

public abstract class FastNettyWebSocketServer implements WebSocketServer {
    private Channel serverChannel;
    private NettyWebsocketServerConfig config;

    public void start(NettyWebsocketServerConfig config) {
        this.config = config;
        ServerBootstrap server = new ServerBootstrap();
        server.group(config.getBoosGroup(), config.getWorkerGroup());
        server.channel(NioServerSocketChannel.class);
        server.childHandler(config.getChildHandler());

        ChannelFuture future = server.bind(config.getPort());
        future.addListener(f -> {
            if (f.isDone() && f.isSuccess()) {
                this.serverChannel = future.channel();
                log.info("Start ws server success");
                log.info("boos group thread number {}", config.getBoosGroup().executorCount());
                log.info("worker group thread number {}", config.getWorkerGroup().executorCount());
            }
            if (f.isDone() && f.cause() != null) {
                log.error("Start ws server fail throw={}", f.cause().getMessage());
                future.channel().close();
            }
        });
    }

    public void start(final int port) {
        start(new NettyWebsocketServerConfig() {
            @Override
            public ChannelHandler getChildHandler() {
                return new WebSocketChannelInitializer(FastNettyWebSocketServer.this);
            }

            @Override
            public int getPort() {
                return port;
            }
        });
    }

    public void start() {
        start(8080);
    }

    public void stop() {
        if (serverChannel != null && serverChannel.isOpen()) {
            final int waitSec = 10;
            CountDownLatch latch = new CountDownLatch(1);
            serverChannel.close().addListener(f -> {
                config.getWorkerGroup().schedule(() -> {
                    log.info("Shutdown dispatcher success...");
                    config.getWorkerGroup().shutdownGracefully();
                    latch.countDown();
                }, waitSec - 2, TimeUnit.SECONDS);

                log.info("Close ws server socket success={}", f.isSuccess());
                config.getBoosGroup().shutdownGracefully();
            });

            try {
                boolean flag = latch.await(waitSec, TimeUnit.SECONDS);
                if(!flag){
                    log.warn("Shutdown ws server timeout");
                }
            } catch (InterruptedException e) {
                log.warn("Shutdown ws server interrupted exception={}", e.getMessage());
            }
        }
    }
}

默认的通道实现

public class  WebSocketChannelInitializer extends ChannelInitializer {

    private final WebSocketServer server;

    public WebSocketChannelInitializer(WebSocketServer server){
        this.server = server;
    }

    @Override
    protected void initChannel(SocketChannel ch) {
        //二进制流在通道中被处理
        ChannelPipeline pipeline = ch.pipeline();
        // HttpRequestDecoder和HttpResponseEncoder的一个组合，针对http协议进行编解码
        pipeline.addLast("httpServerCodec", new HttpServerCodec());//设置解码器
        //分块向客户端写数据，防止发送大文件时导致内存溢出， channel.write(new ChunkedFile(new File("bigFile.mkv")))
        pipeline.addLast(new ChunkedWriteHandler());//用于大数据的分区传输
        // 将HttpMessage和HttpContents聚合到一个完成的 FullHttpRequest或FullHttpResponse中
        // 具体是FullHttpRequest对象还是FullHttpResponse对象取决于是请求还是响应
        // 需要放到HttpServerCodec这个处理器后面
        pipeline.addLast(new HttpObjectAggregator(1024 * 2));//聚合器,使用websocket会用到
        // webSocket 数据压缩扩展，当添加这个的时候WebSocketServerProtocolHandler的第三个参数需要设置成true
        pipeline.addLast(new WebSocketServerCompressionHandler());
        // 服务器端向外暴露的 web socket 端点，当客户端传递比较大的对象时，maxframeSize参数的值需要调大
        pipeline.addLast(new WebSocketServerAuthProtocolHandler("/", null, true, 65536,server));
        pipeline.addLast(new LengthFieldPrepender(4));
        // 业务代码
        pipeline.addLast(new WebSocketServerChannelInboundHandler(server));
    }
}

提供session支持

public class WebSocketServerChannelInboundHandler extends SimpleChannelInboundHandler