Netty如何實現(xiàn)消息推送

Netty 是一個利用 Java 的高級網(wǎng)絡的能力，隱藏其背后的復雜性而提供一個易于使用的 API 的客戶端/服務器框架。

Maven依賴

io.netty netty-all 4.1.36.Final

SpringBootApplication

啟動器中需要new一個NettyServer，并顯式調(diào)用啟動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)//綁定監(jiān)聽端口 .childHandler(newChannelInitializer(){//綁定客戶端連接時候觸發(fā)操作 @Override protectedvoidinitChannel(SocketChannelch)throwsException{ System.out.println("收到新連接"); //websocket協(xié)議本身是基于http協(xié)議的，所以這邊也要使用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();//服務器異步創(chuàng)建綁定 System.out.println(NettyServer.class+"啟動正在監(jiān)聽："+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客戶端發(fā)送消息處理

/** *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)//綁定監(jiān)聽端口 .childHandler(newChannelInitializer(){//綁定客戶端連接時候觸發(fā)操作 @Override protectedvoidinitChannel(SocketChannelch)throwsException{ System.out.println("收到新連接"); //websocket協(xié)議本身是基于http協(xié)議的，所以這邊也要使用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();//服務器異步創(chuàng)建綁定 System.out.println(NettyServer.class+"啟動正在監(jiān)聽："+cf.channel().localAddress()); cf.channel().closeFuture().sync();//關閉服務器通道 }finally{ group.shutdownGracefully().sync();//釋放線程池資源 bossGroup.shutdownGracefully().sync(); } } }

socket.html

主要是連接ws，發(fā)送消息，以及消息反饋

//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協(xié)議！"); } 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參數(shù)

1.首先，調(diào)整一下加載handler的順序，優(yōu)先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，處理參數(shù)byzhengkai.blog.csdn.net if(null!=msg&&msginstanceofFullHttpRequest){ FullHttpRequestrequest=(FullHttpRequest)msg; Stringuri=request.uri(); MapparamMap=getUrlParams(uri); System.out.println("接收到的參數(shù)是："+JSON.toJSONString(paramMap)); //如果url包含參數(shù)，需要處理 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("客戶端收到服務器數(shù)據(jù)："+frame.text()); sendAllMessage(frame.text()); } super.channelRead(ctx,msg); } @Override protectedvoidchannelRead0(ChannelHandlerContextchannelHandlerContext,TextWebSocketFrametextWebSocketFrame)throwsException{ } privatevoidsendAllMessage(Stringmessage){ //收到信息后，群發(fā)給所有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.改造后控制臺輸出情況

收到新連接 與客戶端建立連接，通道開啟！ 接收到的參數(shù)是：{"uid":"666","gid":"777"} /ws 客戶端收到服務器數(shù)據(jù)：142531:這里輸入消息 客戶端收到服務器數(shù)據(jù)：142531:這里輸入消息 客戶端收到服務器數(shù)據(jù)：142531:這里輸入消息

failed: WebSocket opening handshake timed out

聽說是ssl wss的情況下才會出現(xiàn)，來自 @around-gao 的解決方法： 把MyWebSocketHandler和WebSocketServerProtocolHandler調(diào)下順序就好了。

審核編輯：彭靜