MINA2实用手册.docx

mina2入门mina框架是对java的nio包的一个封装，简化了nio程序开发的难度，封装了很多底层的细节，然开发者把精力集中到业务逻辑上来，最近做了一个相关的项目，为了备忘对mina做一个总结。一、 服务端初始化及参数配置mina2初始化很简单。基本的初始化参数如下：/初始化acceptor可以不指定线程数量，mina2里面默认是cpu数量+2 niosocketacceptor acceptor = new niosocketacceptor(5); java.util.concurrent.executor threadpool = executors.newfixedthreadpool(1500);/建立线程池 /加入过滤器（filter）到acceptor acceptor.getfilterchain().addlast(exector, new executorfilter(threadpool); /编码解码器acceptor.getfilterchain().addlast(codec, new protocolcodecfilter(new webdecoder()，new xmlencoder(); /日志 loggingfilter filter = new loggingfilter(); filter.setexceptioncaughtloglevel(loglevel.debug); filter.setmessagereceivedloglevel(loglevel.debug); filter.setmessagesentloglevel(loglevel.debug); filter.setsessionclosedloglevel(loglevel.debug); filter.setsessioncreatedloglevel(loglevel.debug); filter.setsessionidleloglevel(loglevel.debug); filter.setsessionopenedloglevel(loglevel.debug); acceptor.getfilterchain().addlast(logger, filter); acceptor.setreuseaddress(true);/设置的是主服务监听的端口可以重用 acceptor.getsessionconfig().setreuseaddress(true);/设置每一个非主监听连接的端口可以重用 mina2中，当启动一个服务端的时候，要设定初始化缓冲区的长度，如果不设置这个值，系统默认为2048，当客户端发过来的消息超过设定值的时候，mina2的机制是分段接受的，将字符是放入缓冲区中读取，所以在读取消息的时候，需要判断有多少次。这样的好处就是可以节省通讯的流量。acceptor.getsessionconfig().setreceivebuffersize(1024);/设置输入缓冲区的大小 acceptor.getsessionconfig().setsendbuffersize(10240);/设置输出缓冲区的大小 /设置为非延迟发送，为true则不组装成大包发送，收到东西马上发出 acceptor.getsessionconfig().settcpnodelay(true); /设置主服务监听端口的监听队列的最大值为100，如果当前已经有100个连接，再新的连接来将被服务器拒绝 acceptor.setbacklog(100); acceptor.setdefaultlocaladdress(new inetsocketaddress(port); /加入处理器（handler）到acceptor acceptor.sethandler(new yourhandler(); acceptor.bind(); 二、 初始化客户端客户端的初始化和服务器端其实是一样的，就是初始化类不一样，客户端是作为发送者的socketconnector connector = new niosocketconnector();connector.getfilterchain().addlast(codec,new protocolcodecfilter(new xmlcodecfactory(charset1. .forname(charsetname), null, serttype);/指定线程池connector.getfilterchain().addlast(executor, new 、executorfilter(); /指定业务处理类connector.sethandler(this);三、 处理流程niosocketacceptor是mina的适配器，一切都是从这里开始的。mina中有个过滤器和处理器的概念，过滤器用来过滤数据，处理器用来处理数据。具体来说mina的处理模型就是request-过滤器a-过滤器b-处理器-过滤器b-过滤器a-response,这里的request和response类似serlvet的request和response。acceptor.getfilterchain().addlast(codec, new protocolcodecfilter(new webdecoder()，new xmlencoder(); /request-webdecoder-xmlhander-webencode-response acceptor.getfilterchain().addlast(codec, new protocolcodecfilter(new webdecoder()，new xmlencoder();/这里是处理逻辑的关键部位，请求的处理都是在webdecoder类和xmlencoder类中处理，可以明显从命名上看出来一个是用来解码，另一个是用来编码，requet过来后先进入webdecoder类(实现了protocoldecoder接口)进行解码处理，这里可以加入自己的逻辑把传进来的流解码成自己需要的信息。而xmlencoder类(实现了protocolencoder接口)是进行编码，在这个类里面加入自己的逻辑把处理后的信息组装发送给客户端(response)。而在解码和编码过程中xmlhander(扩展了iohandleradapter抽象类)起到了处理器的作用。现在详细描述一下request-webdecoder-xmlhander-webencode-response的过程：客户端发送一个请求到mina服务器，这里相当于来了一个requet。请求首先来到webdecoder类(实现了protocoldecoder接口)中的 boolean decode(iosession session, iobuffer in, protocoldecoderoutput out) throws exception方法 /* 参数in：用户请求信息全存在这里，读数据就从in这里读。 参数out：用来输出处理后的数据到filter的下一个过滤器，如果没有过滤器了就输出到xmlhander，这里有点和 servelt的过滤器类似。利用out.write(object object);这个函数可以把数据传到下一个filter。我们可以自己定义一个对象，我们假设为request，用它来传递消息，那末这里就可以写成out.write(new requsetmessage(); 如果这个方法返回false，就是说当前逻辑包还没接收完（也就是当前的iobuffer并没有包含足够的数据），需要再次 执行decode方法（再次获取新的iobuffer），用来获取足够的数据。如果返回值为true就表示可以不执行decode方 法了，但是要激活handler方法，必须要调用out.write方法。 public class requestmessage/这里什么也不做 */ */然后到xmlhander(扩展了iohandleradapter抽象类)中的 void messagereceived(iosession session, object message) throws exception方法 writefuture future = session.write(response);/session中必须加入这个代码，才会激活encode方法 future.addlistener(iofuturelistener.close);/这个的作用是发送完毕后关闭连接，加了就是短连接，不然是长连接;在xmlhanler类中可以在重载sessionidle方法，这个方法判断整个socket连接通道是否空闲，可以再这里间隔（在服务店启动的时候设置idletime）发送心跳包来保持各个长连接：/* *当网络通道空闲时此方法被调用，在这里可以判断是读空闲、写空闲还是两个都空闲，以便做出正确的处理 一般的网络通讯程序都要与服务器端保持长连接，所以这里可以发一下网络测试数据以保持与服务器端的连接 * param session 会话信息 * param status 状态 * throws exception 异常 */ overridepublic void sessionidle(iosession session, idlestatus status) throws exceptioniofuturelistener里面有个operationcomplete(iofuture future)方法，当流发送完成之后才调用这个方法。 /* 参数message：用来获取filter传递过来的对象.对应代码requestmessage request = (requestmessage) message; 参数session:用来发送数据到filter.对应代码session.write(new responsemessage(); public class responsemessage/这里什么也不做,假设存放处理后的数据 注意：对于一个mina程序而言，对于xmlhander类只生成一个对象，所以要考虑线程安全问题 */然后到xmlencoder类(实现了protocolencoder接口)中的 boolean encode(iosession session, object message, protocolencoderoutput out) throws exception 方法 /* 参数message：用来获取上一个filter节点的数据或者处理器的数据（如果这个过滤器为最靠近处理器的那个） responsemessage response = (responsemessage)message; 参数out：用来输出数据到下一个filter节点过或者到客户端,用out.write(object encodedmessage)把数据发送 出去，但是要注意的是，如果这个filter下一个节点如果是客户端的话，那个这个encodedmessage数据必须为 iobuffer类型的，可以利用iobuffer.wrap(byte bytearray)这个方法来格式化输出数据 */四、 大容量包的处理mina2中（mina2 rc版本，mina2.0正式版已经发布)服务端接受数据默认有一定长度的缓冲区（可以在启动的时候设置)。那么对于大报文，怎么处理呢？比如说超过1024，甚至更多？mina2为了节省网络流量，提高处理效率，会将大报文自动拆分（可能是存放mina2中的缓冲区里面)：比如2048字节的报文，就会拆分成两次；那么在接受的时候，就有一个如何判断是完整报文的问题，或者说是一个拆包组包的问题。 mina2中初始化服务的时候是可以设置输入和输出的缓冲区的： acceptor.getsessionconfig().setreadbuffersize(1024);a) acceptor.getsessionconfig().setreadbuffersize(1024); mina2提供的案例是，在iosession中设置一个类似于session，存在在当前iosession中的全局变量，在此iosession中有效。 privatefinalattributekeytest=newattributekey(getclass(),test); 五、 private final attributekey test = new attributekey(getclass(), test); 大家都知道,通过 socket tcp/ip传输过来的报文是不知道边界的，所以一般会约定在前端固定长度的字节加上报文长度，让server来根据这个长度来确定整个报文的边界，在我前面的博文有提到。其实mina2中有： prefixeddataavailable(4) int 方法，来判断固定长度的报文长度，但是参数只能是1，2，4；该方法很好用。判断前四字节的整型值是否大于等于整个缓冲区的数据。可以方便的判断一次 messagereceived 过来的数据是否完整。（前提是自己设计的网络通讯协议前四字节等于发送数据的长度） ，如果你不是设定1，2，4字节来作为长度的话，那么就没辙了。 在你的解码操作中，mina2的缓冲区发多少次报文，你的decode方法就会调用多少次。 上面设置了session之后，可以采用一个方法： /* * *paramsession *会话信息 *return返回session中的累积 */privatecontextgetcontext(iosessionsession) contextctx=(context)session.getattribute(context); if(ctx=null) ctx=newcontext(); session.setattribute(context,ctx); returnctx; /* param session会话信息return 返回session中的累积*/private context getcontext(iosession session) context ctx = (context) session.getattribute(context);if (ctx = null) ctx = new context();session.setattribute(context, ctx);return ctx;然后在你的decode方法中，首先从session取出数据对象，进行拼接： contextctx=getcontext(session); /先把当前buffer中的数据追加到context的buffer当中 ctx.append(iobuffer); /把position指向0位置，把limit指向原来的position位置 iobufferbuf=ctx.getbuffer(); buf.flip();context ctx = getcontext(session);/ 先把当前buffer中的数据追加到context的buffer当中ctx.append(iobuffer);/ 把position指向0位置，把limit指向原来的position位置iobuffer buf = ctx.getbuffer();buf.flip(); 接着读取每次报文的总长度： /读取消息头部分 bytebleng=newbytepackheadlength; buf.get(bleng); intlength=-1; try length=integer.parseint(newstring(bleng); catch(numberformatexceptionex) ex.printstacktrace(); if(length0) ctx.setmsglength(length); / 读取消息头部分byte bleng = new bytepackheadlength;buf.get(bleng);int length = -1;try length = integer.parseint(new string(bleng); catch (numberformatexception ex) ex.printstacktrace();if (length 0) ctx.setmsglength(length);在读取到每次报文的长度之后，就接着循环判断buf里面的字节数据是否已经全部接受完毕了，如果没有接受完毕，那么就不处理；下面是完整处理的代码： while(buf.remaining()=packheadlength) buf.mark(); /设置总长度 if(ctx.getmsglength()0) ctx.setmsglength(length); /读取消息头部分 intlength=ctx.getmsglength(); /检查读取的包头是否正常，不正常的话清空buffer if(lengthmaxpacklength2) buf.clear(); out.write(error!); break; /读取正常的消息包，并写入输出流中，以便iohandler进行处理 elseif(lengthpackheadlength&buf.remaining()=length) /完整的数据读取之后，就可以开始做你自己想做的操作了 else /如果消息包不完整 /将指针重新移动消息头的起始位置 buf.reset(); break; if(buf.hasremaining()/如果有剩余的数据，则放入session中 /将数据移到buffer的最前面 iobuffertemp=iobuffer.allocate(2048).setautoexpand( true); temp.put(buf); temp.flip(); buf.clear(); buf.put(temp); else/如果数据已经处理完毕，进行清空 buf.clear(); while (buf.remaining() = packheadlength) buf.mark();/ 设置总长度if (ctx.getmsglength() 0) ctx.setmsglength(length);/ 读取消息头部分int length = ctx.getmsglength();/ 检查读取的包头是否正常，不正常的话清空bufferif (length maxpacklength2) buf.clear();out.write(error!);break;/ 读取正常的消息包，并写入输出流中，以便iohandler进行处理 else if (length packheadlength & buf.remaining() = length) /完整的数据读取之后，就可以开始做你自己想做的操作了 else / 如果消息包不完整/ 将指针重新移动消息头的起始位置buf.reset();break;if (buf.hasremaining() / 如果有剩余的数据，则放入session中/ 将数据移到buffer的最前面iobuffer temp = iobuffer.allocate(2048).setautoexpand(true);temp.put(buf);temp.flip();buf.clear();buf.put(temp); else / 如果数据已经处理完毕，进行清空buf.clear();为了便于操作，最好设置一个内部类： privateclasscontext privatefinalcharsetdecoderdecoder; privateiobufferbuf; privateintmsglength=0; privateintoverflowposition=0; /* * * */privatecontext() decoder=charset.newdecoder(); buf=iobuffer.allocate(80).setautoexpand(true); /* * * *returncharsetdecoder */publiccharsetdecodergetdecoder() returndecoder; /* * * *returniobuffer */publiciobuffergetbuffer() returnbuf; /* * * *returnoverflowposition */publicintgetoverflowposition() returnoverflowposition; /* * * *returnmatchcount */publicintgetmsglength() returnmsglength; /* * * *parammatchcount *报文长度 */publicvoidsetmsglength(intmsglength) this.msglength=msglength; /* * * */publicvoidreset() this.buf.clear(); this.overflowposition=0; this.msglength=0; this.decoder.reset(); /* * *paramin *输入流 */publicvoidappend(iobufferin) getbuffer().put(in); 五 多个socket通讯的处理在mina2中两个socket server进行通讯，可以采用虚拟机内部的管道的方式。在mina2的源码包里面自带了这个例子： ioacceptoracceptor=newvmpipeacceptor(); vmpipeaddressaddress=newvmpipeaddress(8080); /setupserver acceptor.sethandler(newtennisplayer(); acceptor.bind(address); /connecttotheserver. vmpipeconnectorconnector=newvmpipeconnector(); connector.sethandler(newtennisplayer(); connectfuturefuture=connector.connect(address); future.awaituninterruptibly(); iosessionsession=future.getsession(); /sendthefirstpingmessage session.write(newtennisball(10); /waituntilthematchends. session.getclosefuture().awaituninterruptibly(); acceptor.unbind();六、 ioacceptor acceptor = new vmpipeacceptor();1. vmpipeaddress address = new vmpipeaddress(8080);2. / set up server3. acceptor.sethandler(new tennisplayer();4. acceptor.bind(address);5. / connect to the server.6. vmpipeconnector connector = new vmpipeconnector();7. connector.sethandler(new tennisplayer();8. connectfuture future = connector.connect(address);9. future.awaituninterruptibly();10. iosession session = future.getsession();11. / send the first ping message12. session.write(new tennisball(10);13. / wait until the match ends.14. session.getclosefuture().awaituninterruptibly();15. acceptor.unbind();也可以将iosession对方放入全局的线程安全的map中去，当需要发送的时候根据key取出来，然后write出去。六 min2的bug我们知道，在mina2中，发送和接受时两个独立的工作线程，但是可以设置一个参数，当服务端发送消息之后同步读取客户端的返回: session.getconfig().setusereadoperation(true);七、 session.getconfig().setusereadoperation(true);近日，采用mina2（rc)的同步读取方法，发现无法真的同步读取客户端的返回； 场景是：服务端发送一个消息给客户端，需要同步等待客户端的一个消息回执，然后服务端的程序继续执行； 但是实际在使用的时候这个设置无效。 sendsession.getconfig().setusereadoperation(true); writefuturefuture=sendsession.write(xmlmsgbean);/发送数据 future.awaituninterruptibly();/等待发送数据操作完成 if(future.getexception()!=null) thrownewappexception(future.getexception().getmessage(); if(future.iswritten() /数据已经被成功发送 logger.debug(数据已经被成功发送); readfuturereadfuture=sendsession.read(); readfuture.awaituninterruptibly(); if(readfuture.getexception()!=null) thrownewappexception(readfuture.getexception().getmessage(); sendsession.getconfig().setusereadoperation(false); return(xmlmsgbean)readfuture.getmessage().getstrerrmsg(); else /数据发送失败 logger.debug(数据发送失败); 后来用google搜索了一下，发现在mina的官网上，老外同样问了一个一模一样的问题，并且提了一个bug上去，但是目前bug的状态还是open； /jira/browse/dirmina-777 imattemptingtoperformasynchronouswrite/readinademux-basedclientapplicationwithmina2.0rc1,butitseemstogetstuck.hereismycode: code publicbooleanlogin(finalstringusername,finalstringpassword) /blockinboundmessages session.getconfig().setusereadoperation(true); /sendtheloginrequest finalloginrequestloginrequest=newloginrequest(username,password); finalwritefuturewritefuture=session.write(loginrequest); writefuture.awaituninterruptibly(); if(writefuture.getexception()!=null) session.getconfig().setusereadoperation(false); returnfalse; /retrievetheloginresponse finalreadfuturereadfuture=session.read(); readfuture.awaituninterruptibly(); if(readfuture.getexception()!=null) session.getconfig().setusereadoperation(false); returnfalse; /stopblockinginboundmessages session.getconfig().setusereadoperation(false); /determineifthelogininfoprovidedwasvalid finalloginresponseloginresponse=(loginresponse)readfuture.getmessage(); returnloginresponse.getsuccess(); code icanseeontheserversidethattheloginrequestobjectisretrieved,andaloginresponsemessageissent.ontheclientside,thedemuxingprotocolcodecfactoryreceivestheresponse,butafterthrowinginsomelogging,icanseethattheclientgetsstuckonthecalltoreadfuture.awaituninterruptibly(). icantforthelifeofmefigureoutwhyitisstuckherebaseduponmyowncode.iproperlysetthereadoperationtotrueonthesessionconfig,meaningthatmessagesshouldbeblocked.however,itseemsasifthemessagenolongerexistsbytimeitrytoreadresponsemessagessynchronously.im attempting to perform a synchronous write/read in a demux-based client application with mina 2.0 rc1, but it seems to get stuck. here is my code: code public boolean login(final string username, final string password) / block inbound messages session.getconfig().setusereadoperation(true); / send the login reque