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

package com.wangjing.socket.server;
 
import com.wangjing.socket.handler.CoordinationSocketHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
 
public class CoordinationNettyServer {
 
    private final int port;
 
    public CoordinationNettyServer(int port) {
        this.port = port;
    }
 
 
    public void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap sb = new ServerBootstrap();
            sb.option(ChannelOption.SO_BACKLOG, 1024);
            sb.group(group, bossGroup) // 綁定線程池
                    .channel(NioServerSocketChannel.class) // 指定使用的channel
                    .localAddress(this.port)// 綁定監(jiān)聽端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 綁定客戶端連接時候觸發(fā)操作
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //websocket協(xié)議本身是基于http協(xié)議的，所以這邊也要使用http解編碼器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //以塊的方式來寫的處理器
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            ch.pipeline().addLast(new HttpObjectAggregator(8192));
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
                            ch.pipeline().addLast(new CoordinationSocketHandler());//自定義消息處理類
                        }
                    });
            ChannelFuture cf = sb.bind().sync(); // 服務器異步創(chuàng)建綁定
            System.out.println(CoordinationNettyServer.class + "已啟動，正在監(jiān)聽： " + cf.channel().localAddress());
            cf.channel().closeFuture().sync(); // 關(guān)閉服務器通道
        } finally {
            group.shutdownGracefully().sync(); // 釋放線程池資源
            bossGroup.shutdownGracefully().sync();
        }
    }
}
 

package com.wangjing.socket.handler;
 
import com.wangjing.socket.pool.CoordinationChannelHandlerPool;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 
public class CoordinationSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
 
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("與客戶端建立連接，通道開啟！");
        //添加到channelGroup通道組
        CoordinationChannelHandlerPool.channelGroup.add(ctx.channel());
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("與客戶端斷開連接，通道關(guān)閉！");
        //從channelGroup通道組刪除
        CoordinationChannelHandlerPool.channelGroup.remove(ctx.channel());
    }
 
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //接收的消息
        System.out.println(String.format("收到客戶端%s的數(shù)據(jù)：%s", ctx.channel().id(), msg.text()));
 
        // 單獨發(fā)消息
        // sendMessage(ctx);
        // 群發(fā)消息
        sendAllMessage();
    }
 
    private void sendMessage(ChannelHandlerContext ctx) throws InterruptedException {
        String message = "我是服務器，你好呀";
        ctx.writeAndFlush(new TextWebSocketFrame("hello"));
    }
 
    private void sendAllMessage() {
        String message = "我是服務器，這是群發(fā)消息";
        CoordinationChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message));
    }
 
 
}

package com.wangjing.socket.pool;
 
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
 
public class CoordinationChannelHandlerPool {
 
    public CoordinationChannelHandlerPool() {
    }
 
    //可以存儲userId與ChannelId的映射表
//    public static ConcurrentHashMap<String, ChannelId> channelIdMap = new ConcurrentHashMap<>();
 
    //channelGroup通道組
    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 
}

package com.wangjing.socket;
 
import com.wangjing.socket.server.CoordinationNettyServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication(scanBasePackages = "com.wangjing")
public class SocketApplication {
 
 
    public static void main(String[] args) {
        SpringApplication.run(SocketApplication.class, args);
 
        try {
            new CoordinationNettyServer(8804).start();
        } catch (Exception e) {
            System.out.println("NettyServerError:" + e.getMessage());
        }
    }
 
}