Netty如何实现消息推送

Netty 是一个利用 Java 的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。

Maven依赖

io.netty netty-all 4.1.36.Final

SpringBootApplication

启动器中需要new一个NettyServer，并显式调用启动netty。

@SpringBootApplication publicclassSpringCloudStudyDemoApplication{ publicstaticvoidmain(String[]args){ SpringApplication.run(SpringCloudStudyDemoApplication.class,args); try{ newNettyServer(12345).start(); System.out.println("https://blog.csdn.net/moshowgame"); System.out.println("http://127.0.0.1:6688/netty-websocket/index"); }catch(Exceptione){ System.out.println("NettyServerError:"+e.getMessage()); } } }

NettyServer

启动的NettyServer，这里进行配置

/** *NettyServerNetty服务器配置 *@authorzhengkai.blog.csdn.net *@date2019-06-12 */ publicclassNettyServer{ privatefinalintport; publicNettyServer(intport){ this.port=port; } publicvoidstart()throwsException{ EventLoopGroupbossGroup=newNioEventLoopGroup(); EventLoopGroupgroup=newNioEventLoopGroup(); try{ ServerBootstrapsb=newServerBootstrap(); sb.option(ChannelOption.SO_BACKLOG,1024); sb.group(group,bossGroup)//绑定线程池 .channel(NioServerSocketChannel.class)//指定使用的channel .localAddress(this.port)//绑定监听端口 .childHandler(newChannelInitializer(){//绑定客户端连接时候触发操作 @Override protectedvoidinitChannel(SocketChannelch)throwsException{ System.out.println("收到新连接"); //websocket协议本身是基于http协议的，所以这边也要使用http解编码器 ch.pipeline().addLast(newHttpServerCodec()); //以块的方式来写的处理器 ch.pipeline().addLast(newChunkedWriteHandler()); ch.pipeline().addLast(newHttpObjectAggregator(8192)); ch.pipeline().addLast(newWebSocketServerProtocolHandler("/ws",null,true,65536*10)); ch.pipeline().addLast(newMyWebSocketHandler()); } }); ChannelFuturecf=sb.bind().sync();//服务器异步创建绑定 System.out.println(NettyServer.class+"启动正在监听："+cf.channel().localAddress()); cf.channel().closeFuture().sync();//关闭服务器通道 }finally{ group.shutdownGracefully().sync();//释放线程池资源 bossGroup.shutdownGracefully().sync(); } } }

MyChannelHandlerPool

通道组池，管理所有websocket连接

/** *MyChannelHandlerPool *通道组池，管理所有websocket连接 *@authorzhengkai.blog.csdn.net *@date2019-06-12 */ publicclassMyChannelHandlerPool{ publicMyChannelHandlerPool(){} publicstaticChannelGroupchannelGroup=newDefaultChannelGroup(GlobalEventExecutor.INSTANCE); }

MyWebSocketHandler

处理ws一下几种情况：

channelActive与客户端建立连接

channelInactive与客户端断开连接

channelRead0客户端发送消息处理

/** *NettyServerNetty服务器配置 *@authorzhengkai.blog.csdn.net *@date2019-06-12 */ publicclassNettyServer{ privatefinalintport; publicNettyServer(intport){ this.port=port; } publicvoidstart()throwsException{ EventLoopGroupbossGroup=newNioEventLoopGroup(); EventLoopGroupgroup=newNioEventLoopGroup(); try{ ServerBootstrapsb=newServerBootstrap(); sb.option(ChannelOption.SO_BACKLOG,1024); sb.group(group,bossGroup)//绑定线程池 .channel(NioServerSocketChannel.class)//指定使用的channel .localAddress(this.port)//绑定监听端口 .childHandler(newChannelInitializer(){//绑定客户端连接时候触发操作 @Override protectedvoidinitChannel(SocketChannelch)throwsException{ System.out.println("收到新连接"); //websocket协议本身是基于http协议的，所以这边也要使用http解编码器 ch.pipeline().addLast(newHttpServerCodec()); //以块的方式来写的处理器 ch.pipeline().addLast(newChunkedWriteHandler()); ch.pipeline().addLast(newHttpObjectAggregator(8192)); ch.pipeline().addLast(newWebSocketServerProtocolHandler("/ws","WebSocket",true,65536*10)); ch.pipeline().addLast(newMyWebSocketHandler()); } }); ChannelFuturecf=sb.bind().sync();//服务器异步创建绑定 System.out.println(NettyServer.class+"启动正在监听："+cf.channel().localAddress()); cf.channel().closeFuture().sync();//关闭服务器通道 }finally{ group.shutdownGracefully().sync();//释放线程池资源 bossGroup.shutdownGracefully().sync(); } } }

socket.html

主要是连接ws，发送消息，以及消息反馈

//byzhengkai.blog.csdn.net varsocket; if(!window.WebSocket){ window.WebSocket=window.MozWebSocket; } if(window.WebSocket){ socket=newWebSocket("ws://127.0.0.1:12345/ws"); socket.onmessage=function(event){ varta=document.getElementById('responseText'); ta.value+=event.data+" "; }; socket.onopen=function(event){ varta=document.getElementById('responseText'); ta.value="Netty-WebSocket服务器。。。。。。连接 "; }; socket.onclose=function(event){ varta=document.getElementById('responseText'); ta.value="Netty-WebSocket服务器。。。。。。关闭 "; }; }else{ alert("您的浏览器不支持WebSocket协议！"); } functionsend(message){ if(!window.WebSocket){return;} if(socket.readyState==WebSocket.OPEN){ socket.send(message); }else{ alert("WebSocket 连接没有建立成功！"); } } ID TEXT

服务端返回的应答消息

Controller

写好了html当然还需要一个controller来引导页面。

@RestController publicclassIndexController{ @GetMapping("/index") publicModelAndViewindex(){ ModelAndViewmav=newModelAndView("socket"); mav.addObject("uid",RandomUtil.randomNumbers(6)); returnmav; } }

效果演示

改造netty支持url参数

1.首先，调整一下加载handler的顺序，优先MyWebSocketHandler在WebSocketServerProtocolHandler之上。

ch.pipeline().addLast(newMyWebSocketHandler()); ch.pipeline().addLast(newWebSocketServerProtocolHandler("/ws",null,true,65536*10)); 2.其次，改造MyWebSocketHandler的channelRead方法，首次连接会是一个FullHttpRequest类型，可以通过FullHttpRequest.uri()获取完整ws的URL地址，之后接受信息的话，会是一个TextWebSocketFrame类型。

publicclassMyWebSocketHandlerextendsSimpleChannelInboundHandler{ @Override publicvoidchannelActive(ChannelHandlerContextctx)throwsException{ System.out.println("与客户端建立连接，通道开启！"); //添加到channelGroup通道组 MyChannelHandlerPool.channelGroup.add(ctx.channel()); } @Override publicvoidchannelInactive(ChannelHandlerContextctx)throwsException{ System.out.println("与客户端断开连接，通道关闭！"); //添加到channelGroup通道组 MyChannelHandlerPool.channelGroup.remove(ctx.channel()); } @Override publicvoidchannelRead(ChannelHandlerContextctx,Objectmsg)throwsException{ //首次连接是FullHttpRequest，处理参数byzhengkai.blog.csdn.net if(null!=msg&&msginstanceofFullHttpRequest){ FullHttpRequestrequest=(FullHttpRequest)msg; Stringuri=request.uri(); MapparamMap=getUrlParams(uri); System.out.println("接收到的参数是："+JSON.toJSONString(paramMap)); //如果url包含参数，需要处理 if(uri.contains("?")){ StringnewUri=uri.substring(0,uri.indexOf("?")); System.out.println(newUri); request.setUri(newUri); } }elseif(msginstanceofTextWebSocketFrame){ //正常的TEXT消息类型 TextWebSocketFrameframe=(TextWebSocketFrame)msg; System.out.println("客户端收到服务器数据："+frame.text()); sendAllMessage(frame.text()); } super.channelRead(ctx,msg); } @Override protectedvoidchannelRead0(ChannelHandlerContextchannelHandlerContext,TextWebSocketFrametextWebSocketFrame)throwsException{ } privatevoidsendAllMessage(Stringmessage){ //收到信息后，群发给所有channel MyChannelHandlerPool.channelGroup.writeAndFlush(newTextWebSocketFrame(message)); } privatestaticMapgetUrlParams(Stringurl){ Mapmap=newHashMap<>(); url=url.replace("?",";"); if(!url.contains(";")){ returnmap; } if(url.split(";").length>0){ String[]arr=url.split(";")[1].split("&"); for(Strings:arr){ Stringkey=s.split("=")[0]; Stringvalue=s.split("=")[1]; map.put(key,value); } returnmap; }else{ returnmap; } } } 3.html中的ws地址也进行改造

socket=newWebSocket("ws://127.0.0.1:12345/ws?uid=666&gid=777"); 4.改造后控制台输出情况

收到新连接 与客户端建立连接，通道开启！ 接收到的参数是：{"uid":"666","gid":"777"} /ws 客户端收到服务器数据：142531:这里输入消息 客户端收到服务器数据：142531:这里输入消息 客户端收到服务器数据：142531:这里输入消息

failed: WebSocket opening handshake timed out

听说是ssl wss的情况下才会出现，来自 @around-gao 的解决方法： 把MyWebSocketHandler和WebSocketServerProtocolHandler调下顺序就好了。

审核编辑：彭静