<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>5.0.0.Alpha2</version></dependency>

@ComponentpublicclassNettyServerRunnerimplementsCommandLineRunner {
privatefinalLoggerLOGGER=LogUtils.getLogger(NettyServerRunner.class);
@Overridepublicvoidrun(String... args) throwsException {
//要监听的端口intport=18000;
// socket地址Stringuri="/ws/api/datapush";
EventLoopGroupbossGroup=newNioEventLoopGroup(1);
EventLoopGroupworkerGroup=newNioEventLoopGroup();
try {
ServerBootstrapbootstrap=newServerBootstrap();
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// 绑定主线程组和工作线程组bootstrap.group(bossGroup, workerGroup) 
    .channel(NioServerSocketChannel.class)
    .handler(newLoggingHandler(LogLevel.INFO))
// 绑定监听端口    .localAddress(port)
// 绑定客户端连接时候触发操作    .childHandler(newMyChannelInitializer(uri))
    .childOption(ChannelOption.TCP_NODELAY, true)
             .childOption(ChannelOption.SO_KEEPALIVE, true);
// 服务器异步创建绑定ChannelFuturecf=bootstrap.bind().sync(); 
LOGGER.warn("启动正在监听： "+cf.channel().localAddress());
// 关闭服务器通道cf.channel().closeFuture().sync();
    } finally {
// 释放线程池资源workerGroup.shutdownGracefully().sync();
bossGroup.shutdownGracefully().sync();
    }
 }

publicclassMyChannelInitializerextendsChannelInitializer<SocketChannel> {
privatestaticfinalintMAX_CONTENT_LENGTH=1024*1024;
privatestaticfinalCharsetCHARSET_UTF8=Charset.forName(CenterKeys.CHARSET_UTF8);
privateStringwsUri;
publicMyChannelInitializer(StringwsUri) {
Preconditions.checkArgument(StringUtils.isNotBlank(wsUri), "Socket地址不能为空");
this.wsUri=wsUri;
    }
@OverrideprotectedvoidinitChannel(SocketChannelchannel) throwsException    {
ChannelPipelinepipeline=channel.pipeline();
// HTTP 服务的解码器pipeline.addLast(newHttpServerCodec());
// HTTP 消息的合并处理pipeline.addLast(newHttpObjectAggregator(MAX_CONTENT_LENGTH));
pipeline.addLast(newStringDecoder(CHARSET_UTF8));
pipeline.addLast(newStringEncoder(CHARSET_UTF8));
// 以块的方式来写的处理器pipeline.addLast(newChunkedWriteHandler());
//自定义的业务handlerpipeline.addLast(MyWebSocketHandler.getInstance());
pipeline.addLast(newWebSocketServerProtocolHandler(wsUri, "WebSocket", true));
    }
}

@SharablepublicclassMyWebSocketHandlerextendsSimpleChannelInboundHandler<TextWebSocketFrame> {
privatestaticfinalLoggerLOGGER=LogUtils.getLogger(MyWebSocketHandler.class);
privatestaticfinalChannelGroupCHANNEL_GROUP=newDefaultChannelGroup(GlobalEventExecutor.INSTANCE);
privatestaticfinalMyWebSocketHandlerINSTANCE=newMyWebSocketHandler();
privateMyWebSocketHandler() {}
publicstaticMyWebSocketHandlergetInstance() {
returnINSTANCE;
    }
@OverridepublicvoidchannelRegistered(ChannelHandlerContextctx) throwsException    {
StringclientIp=getClientAddress(ctx);
super.channelRegistered(ctx);
LOGGER.info("=======收到客户端{}注册通知=======", clientIp);
    }
@OverridepublicvoidhandlerAdded(ChannelHandlerContextctx) throwsException    {
StringclientIp=getClientAddress(ctx);
CHANNEL_GROUP.add(ctx.channel());
super.handlerAdded(ctx);
LOGGER.info("=======收到客户端{}加入通知=======", clientIp);
    }
@OverridepublicvoidhandlerRemoved(ChannelHandlerContextctx) throwsException    {
StringclientIp=getClientAddress(ctx);
CHANNEL_GROUP.remove(ctx.channel());
super.handlerRemoved(ctx);
LOGGER.info("=======收到客户端{}离开通知=======", clientIp);
    }
//当有客户端连接时候会被channelActive监听到，//当断开时会被channelInactive监听到，一般在这两个方法中去保存/移除客户端的通道信息@OverridepublicvoidchannelActive(ChannelHandlerContextctx) throwsException    {
StringclientIp=getClientAddress(ctx);
super.channelActive(ctx);
LOGGER.info("=======收到客户端{}在线通知=======", clientIp);
    }
@OverridepublicvoidchannelInactive(ChannelHandlerContextctx) throwsException    {
StringclientIp=getClientAddress(ctx);
CHANNEL_GROUP.remove(ctx.channel());
super.channelInactive(ctx);
LOGGER.info("=======收到客户端{}离线通知=======", clientIp);
    }
@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx, Throwablecause) {
ctx.close();
    }
@OverridepublicvoidchannelRead(ChannelHandlerContextctx, Objectmessage) throwsException    {
if (null!=message&&messageinstanceofTextWebSocketFrame) 
        {
TextWebSocketFrameframe= (TextWebSocketFrame) message;
LOGGER.info("接收到客户端请求参数：{}。", frame.text());
Stringresult="要通知的消息"LOGGER.info("客户端连接个数：{}。", CHANNEL_GROUP.size());
// 收到信息后，群发给所有channelCHANNEL_GROUP.writeAndFlush(newTextWebSocketFrame(result));
        }
super.channelRead(ctx, message);
    }
/*** .推送消息通知到客户端*  PushFuncType  可以定义一个Socket消息推送业务功能类型的枚举类*/publicstaticvoidnotify(PushFuncTypefuncType) throwsException    {
// 收到信息后，群发给所有channelLOGGER.info("客户端连接个数：{}。", CHANNEL_GROUP.size());
CHANNEL_GROUP.writeAndFlush(newTextWebSocketFrame(funcType.toString()));
    }
@OverrideprotectedvoidmessageReceived(ChannelHandlerContextchannelHandlerContext, TextWebSocketFramemessage)
throwsException {
    }
privateStringgetClientAddress(ChannelHandlerContextctx) 
    {
InetSocketAddressinsocket= (InetSocketAddress) ctx.channel().remoteAddress();
returninsocket.getAddress().getHostAddress();
    }
}