netty长连接实例

通过netty实现服务端与客户端的长连接通讯，及心跳检测。

基本思路：netty服务端通过一个Map保存所有连接上来的客户端SocketChannel,客户端的Id作为Map的key。每次服务器端如果要向某个客户端发送消息，只需根据ClientId取出对应的SocketChannel,往里面写入message即可。心跳检测通过IdleEvent事件，定时向服务端放送Ping消息，检测SocketChannel是否终断。

环境JDK1.8 和netty5

以下是具体的代码实现和介绍：

1公共的Share部分(主要包含消息协议类型的定义)

设计消息类型：

?

1 2 3

publicenumMsgType{ PING,ASK,REPLY,LOGIN }

Message基类：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

//必须实现序列,serialVersionUID一定要有,否者在netty消息序列化反序列化会有问题，接收不到消息！！！ publicabstractclassBaseMsgimplementsSerializable{ privatestaticfinallongserialVersionUID=1L; privateMsgTypetype; //必须唯一，否者会出现channel调用混乱 privateStringclientId; //初始化客户端id publicBaseMsg(){ this.clientId=Constants.getClientId(); } publicStringgetClientId(){ returnclientId; } publicvoidsetClientId(StringclientId){ this.clientId=clientId; } publicMsgTypegetType(){ returntype; } publicvoidsetType(MsgTypetype){ this.type=type; } }

常量设置：

?

1 2 3 4 5 6 7 8 9

publicclassConstants{ privatestaticStringclientId; publicstaticStringgetClientId(){ returnclientId; } publicstaticvoidsetClientId(StringclientId){ Constants.clientId=clientId; } }

登录类型消息：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

publicclassLoginMsgextendsBaseMsg{ privateStringuserName; privateStringpassword; publicLoginMsg(){ super(); setType(MsgType.LOGIN); } publicStringgetUserName(){ returnuserName; } publicvoidsetUserName(StringuserName){ this.userName=userName; } publicStringgetPassword(){ returnpassword; } publicvoidsetPassword(Stringpassword){ this.password=password; } }

心跳检测Ping类型消息：

?

1 2 3 4 5 6

publicclassPingMsgextendsBaseMsg{ publicPingMsg(){ super(); setType(MsgType.PING); } }

请求类型消息：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

publicclassAskMsgextendsBaseMsg{ publicAskMsg(){ super(); setType(MsgType.ASK); } privateAskParamsparams; publicAskParamsgetParams(){ returnparams; } publicvoidsetParams(AskParamsparams){ this.params=params; } } //请求类型参数 //必须实现序列化接口 publicclassAskParamsimplementsSerializable{ privatestaticfinallongserialVersionUID=1L; privateStringauth; publicStringgetAuth(){ returnauth; } publicvoidsetAuth(Stringauth){ this.auth=auth; } }

响应类型消息：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46

publicclassReplyMsgextendsBaseMsg{ publicReplyMsg(){ super(); setType(MsgType.REPLY); } privateReplyBodybody; publicReplyBodygetBody(){ returnbody; } publicvoidsetBody(ReplyBodybody){ this.body=body; } } //相应类型body对像 publicclassReplyBodyimplementsSerializable{ privatestaticfinallongserialVersionUID=1L; } publicclassReplyClientBodyextendsReplyBody{ privateStringclientInfo; publicReplyClientBody(StringclientInfo){ this.clientInfo=clientInfo; } publicStringgetClientInfo(){ returnclientInfo; } publicvoidsetClientInfo(StringclientInfo){ this.clientInfo=clientInfo; } } publicclassReplyServerBodyextendsReplyBody{ privateStringserverInfo; publicReplyServerBody(StringserverInfo){ this.serverInfo=serverInfo; } publicStringgetServerInfo(){ returnserverInfo; } publicvoidsetServerInfo(StringserverInfo){ this.serverInfo=serverInfo; } }

2 Server端：主要包含对SocketChannel引用的Map,ChannelHandler的实现和Bootstrap.

Map:

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

publicclassNettyChannelMap{ privatestaticMap<String,SocketChannel>map=newConcurrentHashMap<String,SocketChannel>(); publicstaticvoidadd(StringclientId,SocketChannelsocketChannel){ map.put(clientId,socketChannel); } publicstaticChannelget(StringclientId){ returnmap.get(clientId); } publicstaticvoidremove(SocketChannelsocketChannel){ for(Map.Entryentry:map.entrySet()){ if(entry.getValue()==socketChannel){ map.remove(entry.getKey()); } } } }

Handler

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50

publicclassNettyServerHandlerextendsSimpleChannelInboundHandler<BaseMsg>{ @Override publicvoidchannelInactive(ChannelHandlerContextctx)throwsException{ //channel失效，从Map中移除 NettyChannelMap.remove((SocketChannel)ctx.channel()); } @Override protectedvoidmessageReceived(ChannelHandlerContextchannelHandlerContext,BaseMsgbaseMsg)throwsException{ if(MsgType.LOGIN.equals(baseMsg.getType())){ LoginMsgloginMsg=(LoginMsg)baseMsg; if("robin".equals(loginMsg.getUserName())&&"yao".equals(loginMsg.getPassword())){ //登录成功,把channel存到服务端的map中 NettyChannelMap.add(loginMsg.getClientId(),(SocketChannel)channelHandlerContext.channel()); System.out.println("client"+loginMsg.getClientId()+"登录成功"); } }else{ if(NettyChannelMap.get(baseMsg.getClientId())==null){ //说明未登录，或者连接断了，服务器向客户端发起登录请求，让客户端重新登录 LoginMsgloginMsg=newLoginMsg(); channelHandlerContext.channel().writeAndFlush(loginMsg); } } switch(baseMsg.getType()){ casePING:{ PingMsgpingMsg=(PingMsg)baseMsg; PingMsgreplyPing=newPingMsg(); NettyChannelMap.get(pingMsg.getClientId()).writeAndFlush(replyPing); }break; caseASK:{ //收到客户端的请求 AskMsgaskMsg=(AskMsg)baseMsg; if("authToken".equals(askMsg.getParams().getAuth())){ ReplyServerBodyreplyBody=newReplyServerBody("serverinfo$$$$!!!"); ReplyMsgreplyMsg=newReplyMsg(); replyMsg.setBody(replyBody); NettyChannelMap.get(askMsg.getClientId()).writeAndFlush(replyMsg); } }break; caseREPLY:{ //收到客户端回复 ReplyMsgreplyMsg=(ReplyMsg)baseMsg; ReplyClientBodyclientBody=(ReplyClientBody)replyMsg.getBody(); System.out.println("receiveclientmsg:"+clientBody.getClientInfo()); }break; default:break; } ReferenceCountUtil.release(baseMsg); } }

ServerBootstrap:

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45

publicclassNettyServerBootstrap{ privateintport; privateSocketChannelsocketChannel; publicNettyServerBootstrap(intport)throwsInterruptedException{ this.port=port; bind(); } privatevoidbind()throwsInterruptedException{ EventLoopGroupboss=newNioEventLoopGroup(); EventLoopGroupworker=newNioEventLoopGroup(); ServerBootstrapbootstrap=newServerBootstrap(); bootstrap.group(boss,worker); bootstrap.channel(NioServerSocketChannel.class); bootstrap.option(ChannelOption.SO_BACKLOG,128); //通过NoDelay禁用Nagle,使消息立即发出去，不用等待到一定的数据量才发出去 bootstrap.option(ChannelOption.TCP_NODELAY,true); //保持长连接状态 bootstrap.childOption(ChannelOption.SO_KEEPALIVE,true); bootstrap.childHandler(newChannelInitializer<SocketChannel>(){ @Override protectedvoidinitChannel(SocketChannelsocketChannel)throwsException{ ChannelPipelinep=socketChannel.pipeline(); p.addLast(newObjectEncoder()); p.addLast(newObjectDecoder(ClassResolvers.cacheDisabled(null))); p.addLast(newNettyServerHandler()); } }); ChannelFuturef=bootstrap.bind(port).sync(); if(f.isSuccess()){ System.out.println("serverstart---------------"); } } publicstaticvoidmain(String[]args)throwsInterruptedException{ NettyServerBootstrapbootstrap=newNettyServerBootstrap(9999); while(true){ SocketChannelchannel=(SocketChannel)NettyChannelMap.get("001"); if(channel!=null){ AskMsgaskMsg=newAskMsg(); channel.writeAndFlush(askMsg); } TimeUnit.SECONDS.sleep(5); } } }

3 Client端：包含发起登录，发送心跳，及对应消息处理

handler

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

publicclassNettyClientHandlerextendsSimpleChannelInboundHandler<BaseMsg>{ //利用写空闲发送心跳检测消息 @Override publicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{ if(evtinstanceofIdleStateEvent){ IdleStateEvente=(IdleStateEvent)evt; switch(e.state()){ caseWRITER_IDLE: PingMsgpingMsg=newPingMsg(); ctx.writeAndFlush(pingMsg); System.out.println("sendpingtoserver----------"); break; default: break; } } } @Override protectedvoidmessageReceived(ChannelHandlerContextchannelHandlerContext,BaseMsgbaseMsg)throwsException{ MsgTypemsgType=baseMsg.getType(); switch(msgType){ caseLOGIN:{ //向服务器发起登录 LoginMsgloginMsg=newLoginMsg(); loginMsg.setPassword("yao"); loginMsg.setUserName("robin"); channelHandlerContext.writeAndFlush(loginMsg); }break; casePING:{ System.out.println("receivepingfromserver----------"); }break; caseASK:{ ReplyClientBodyreplyClientBody=newReplyClientBody("clientinfo****!!!"); ReplyMsgreplyMsg=newReplyMsg(); replyMsg.setBody(replyClientBody); channelHandlerContext.writeAndFlush(replyMsg); }break; caseREPLY:{ ReplyMsgreplyMsg=(ReplyMsg)baseMsg; ReplyServerBodyreplyServerBody=(ReplyServerBody)replyMsg.getBody(); System.out.println("receiveclientmsg:"+replyServerBody.getServerInfo()); } default:break; } ReferenceCountUtil.release(msgType); } }

bootstrap

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50

publicclassNettyClientBootstrap{ privateintport; privateStringhost; privateSocketChannelsocketChannel; privatestaticfinalEventExecutorGroupgroup=newDefaultEventExecutorGroup(20); publicNettyClientBootstrap(intport,Stringhost)throwsInterruptedException{ this.port=port; this.host=host; start(); } privatevoidstart()throwsInterruptedException{ EventLoopGroupeventLoopGroup=newNioEventLoopGroup(); Bootstrapbootstrap=newBootstrap(); bootstrap.channel(NioSocketChannel.class); bootstrap.option(ChannelOption.SO_KEEPALIVE,true); bootstrap.group(eventLoopGroup); bootstrap.remoteAddress(host,port); bootstrap.handler(newChannelInitializer<SocketChannel>(){ @Override protectedvoidinitChannel(SocketChannelsocketChannel)throwsException{ socketChannel.pipeline().addLast(newIdleStateHandler(20,10,0)); socketChannel.pipeline().addLast(newObjectEncoder()); socketChannel.pipeline().addLast(newObjectDecoder(ClassResolvers.cacheDisabled(null))); socketChannel.pipeline().addLast(newNettyClientHandler()); } }); ChannelFuturefuture=bootstrap.connect(host,port).sync(); if(future.isSuccess()){ socketChannel=(SocketChannel)future.channel(); System.out.println("connectserver成功---------"); } } publicstaticvoidmain(String[]args)throwsInterruptedException{ Constants.setClientId("001"); NettyClientBootstrapbootstrap=newNettyClientBootstrap(9999,"localhost"); LoginMsgloginMsg=newLoginMsg(); loginMsg.setPassword("yao"); loginMsg.setUserName("robin"); bootstrap.socketChannel.writeAndFlush(loginMsg); while(true){ TimeUnit.SECONDS.sleep(3); AskMsgaskMsg=newAskMsg(); AskParamsaskParams=newAskParams(); askParams.setAuth("authToken"); askMsg.setParams(askParams); bootstrap.socketChannel.writeAndFlush(askMsg); } } }

具体的例子和相应pom.xml 见https://github.com/WangErXiao/ServerClient

转发请注明来源：http://my.oschina.net/robinyao/blog/399060