SpringBoot整合Netty实现WebSocket的示例代码

<!-- 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) // 绑定监听端口

                     .childHandler( new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作

                         @Override

                         protected void initChannel(SocketChannel ch) throws Exception {

                             //websocket协议本身是基于http协议的，所以这边也要使用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(); // 服务器异步创建绑定

             System.out.println(CoordinationNettyServer. class + "已启动，正在监听： " + cf.channel().localAddress());

             cf.channel().closeFuture().sync(); // 关闭服务器通道

         } 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( "与客户端断开连接，通道关闭！" );

         //从channelGroup通道组删除

         CoordinationChannelHandlerPool.channelGroup.remove(ctx.channel());

     }

     @Override

     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {

         //接收的消息

         System.out.println(String.format( "收到客户端%s的数据：%s" , ctx.channel().id(), msg.text()));

         // 单独发消息

         // sendMessage(ctx);

         // 群发消息

         sendAllMessage();

     }

     private void sendMessage(ChannelHandlerContext ctx) throws InterruptedException {

         String message = "我是服务器，你好呀" ;

         ctx.writeAndFlush( new TextWebSocketFrame( "hello" ));

     }

     private void sendAllMessage() {

         String message = "我是服务器，这是群发消息" ;

         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());

         }

     }

}