Netty WebSocket

Netty实现websocket消息推送

pom 依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.23.Final</version>
</dependency>

websocket 处理类

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

public class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    /**
     * channel 存储
     */
    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        Channel incoming = ctx.channel();
        for (Channel channel : channels) {
            if (channel != incoming) {
                channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]" + msg.text()));
            } else {
                channel.writeAndFlush(new TextWebSocketFrame("[you]" + msg.text()));
            }
        }
    }

    /**
     * 用户加入
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        channels.writeAndFlush(new TextWebSocketFrame("[Server] - " + channel.remoteAddress() + "加入"));
        channels.add(channel);
    }

    /**
     * 用户退出
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        channels.writeAndFlush(new TextWebSocketFrame("[Server] - " + channel.remoteAddress() + "离开"));
        // A closed Channel is automatically removed from ChannelGroup,
        // so there is no need to do "channels.remove(ctx.channel());"
    }

    /**
     * 用户连接
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        System.out.println("Client: " + channel.remoteAddress() + "在线");
    }

    /**
     * 用户掉线
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        System.out.println("Client: " + channel.remoteAddress() + "掉线");
    }

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        Channel channel = ctx.channel();
        System.out.println("Client: " + channel.remoteAddress() + "异常");
        cause.printStackTrace();
        ctx.close();
    }

}

初始化类

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new WebSocketFrameHandler());
    }

}

server启动类

import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

public class WebSocketServer {
    private int port;

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

    public void run() throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new WebSocketServerInitializer())
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true)
            ;
            System.out.println("web socket 启动了");
            ChannelFuture future = b.bind(port).sync();
            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
            System.out.println("web socket 关闭了");
        }

    }

    public static void main(String[] args) throws Exception {
        int port;
        if(args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        //延时发送消息（test使用）
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
        scheduledExecutorService.schedule(new Callable<String>() {

            @Override
            public String call() throws Exception{
                System.out.println("发送系统消息");
                WebSocketFrameHandler.channels.writeAndFlush(new TextWebSocketFrame("系统消息，哈哈"));
                return "ok";
            }
        }, 10, TimeUnit.SECONDS);
        //启动服务
        new WebSocketServer(port).run();
        scheduledExecutorService.shutdown();
    }
}

客户端

//浏览器是否支持WebSocket
if(!window.WebSocket) {
    window.WebSocket = window.MozWebSocket;
}
var socket;
//显示消息容器
var msgContent = document.getElementById("message_content");
//需要发送消息input框框
var msg = document.getElementById("msg");
if(window.WebSocket) {
    //连接websocket
    socket = new WebSocket("ws://127.0.0.1:8080/ws")

    //收到消息
    socket.onmessage = function(event) {
        //debugger
        msgContent.innerHTML += "<span>" + event.data + "</span><br>";
    }
    socket.onopen = function(event) {
        console.log("连接开启!" + event)
    }
    socket.onclose = function(event) {
        console.log("连接关闭!" + event)
    }
} else {
    alert("浏览器不支持WebSocket");
}
function send() {
    if(!window.WebSocket) {
        alert("无法发送消息")
        return;
    }
    //发送消息
    if (socket.readyState == WebSocket.OPEN) {
        socket.send(msg.value)
    } else {
        alert("没有建立连接" + socket.readyState);
    }
}