spring + mina 作为客户端解析H2协议的使用总结
直接上代码
1:spring 的配置文件
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:jee="http://www.springframework.org/schema/jee"xmlns:tx="http://www.springframework.org/schema/tx"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop"default-autowire="byName"default-lazy-init="false"xmlns="http://www.springframework.org/schema/beans"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd" ><description>UPM SERVER 配置文件</description><bean class="org.springframework.beans.factory.config.CustomEditorConfigurer" ><property name="customEditors" ><map><entry key="java.net.SocketAddress" ><bean class="org.apache.mina.integration.beans.InetSocketAddressEditor" /></entry></map></property></bean><bean id="executorFilter" class="org.apache.mina.filter.executor.ExecutorFilter" /><bean id="mdcInjectionFilter" class="org.apache.mina.filter.logging.MdcInjectionFilter" ><constructor-arg value="remoteAddress" /></bean><bean id="codecFilter" class="org.apache.mina.filter.codec.ProtocolCodecFilter" ><constructor-arg><bean class="com.newyulong.upm.io.h2.codec.DefaultProtocolFactory" ><property name="protocolEncoder" ref="protocolEncoder" /><property name="protocolDecoder" ref="protocolDecoder" /></bean></constructor-arg></bean><bean id="protocolDecoder" class="com.newyulong.upm.io.h2.codec.ProtocolDecoder" ></bean><bean id="protocolDecoderFactory" class="com.newyulong.upm.io.h2.ProtocolDecodeFactoryBean"><property name="region" value="${upm.region}"></property><property name="protocolDecoderFactorys"><map><entry key="shanxiProtocolDecoderFactory"><ref local="shanxiProtocolFactory"/></entry> <entry key="xingjiangProtocolDecoderFactory"><ref local="xingjiangProtocolDecoderFactory"/></entry> </map></property><property name="regionProtocolDecoderFactorys"><map><entry key="shanxi" value="shanxiProtocolDecoderFactory"></entry><entry key="xinjiang" value="xingjiangProtocolDecoderFactory"></entry></map></property></bean><bean id="xingjiangProtocolDecoderFactory" class="com.newyulong.upm.io.h2.codec.XinJiangProtocolFactory"></bean> <bean id="shanxiProtocolFactory" class="com.newyulong.upm.io.h2.codec.ShanXiProtocolFactory"></bean><bean id="protocolEncoder" class="com.newyulong.upm.io.h2.codec.ProtocolEncoder" /><bean id="loggingFilter" class="org.apache.mina.filter.logging.LoggingFilter" /><bean id="clientFilterChainBuilder" class="org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder" ><property name="filters" ><map><entry key="executor" value-ref="executorFilter" /><entry key="mdcInjectionFilter" value-ref="mdcInjectionFilter" /><!-- 编码解码的过滤器--><entry key="codecFilter" value-ref="codecFilter" /><entry key="loggingFilter" value-ref="loggingFilter" /></map></property></bean><bean id="clientIoHandler" class="com.newyulong.upm.io.h2.handler.ClientIoHandler" /><!-- 此对象就是作为客户端的对象，此对象中注入了过滤器的对象，包含的过滤对象有，解码，和编码的过滤器 --><bean id="upmConnector" class="org.apache.mina.transport.socket.nio.NioSocketConnector" ><property name="connectTimeout" value="8000" /><property name="connectTimeoutMillis" value="8000" /><!-- 对应上面的过滤器对象的集合,主要包括编码和解码的过滤器--><property name="filterChainBuilder" ref="clientFilterChainBuilder" /><constructor-arg index="0" value="6" /><property name="handler" ref="clientIoHandler" /></bean> <!-- DefaultClientSocket类是我们整个系统的入口,里面注入了IoConnector这个 对象NioSocketConnector也就是上面的upmConnector对象--><bean id="upmClient" class="com.newyulong.upm.io.h2.DefaultClientSocket" destroy-method="dispose" ><property name="connector" ref="upmConnector"/><!-- <property name="synchronization" value="true"/> --><constructor-arg index="0" value="${upm.server.ip}" /><constructor-arg index="1" value="${upm.server.port}" /></bean> </beans> com.newyulong.upm.io.h2.DefaultClientSocket类代码入口就是此类的send方法 package com.newyulong.upm.io.h2;import java.util.concurrent.TimeUnit;import org.apache.mina.core.RuntimeIoException; import org.apache.mina.core.future.CloseFuture; import org.apache.mina.core.future.ConnectFuture; import org.apache.mina.core.future.ReadFuture; import org.apache.mina.core.session.IoSession;import com.kms.components.io.ClientSocket; import com.kms.components.io.dto.IMessage; import com.kms.framework.core.logger.ILogger; import com.kms.framework.core.logger.LoggerFactory;public class DefaultClientSocket extends ClientSocket {protected ILogger debugLogger = LoggerFactory.getOperationLogger(getClass());private String hIp;private int hPort;public DefaultClientSocket(String hostIP, int hostPort) {super(hostIP, hostPort);this.hIp = hostIP;this.hPort = hostPort;}public DefaultClientSocket(String hostIP, int hostPort, String localIP,int localPort) {super(hostIP, hostPort, localIP, localPort);}public IMessage send(IMessage msg) {debugLogger.info("Request ip info, hostIP===>" + this.hIp+ " hostPort===>" + this.hPort);// --连接到远程地址ConnectFuture connectFuture = this.getConnector().connect(this.getRemote());connectFuture.awaitUninterruptibly();IoSession session = null;try {session = connectFuture.getSession();session.getConfig().setUseReadOperation(true);} catch (RuntimeIoException rio) {debugLogger.error("Get Session exception", rio);throw new RuntimeException("can't connect to the bss");}// 等待客户端写出数据 session.write(msg).awaitUninterruptibly();IMessage msg1 = null;// 客户端开始读取数据ReadFuture readFuture = session.read();if (readFuture.awaitUninterruptibly(30, TimeUnit.SECONDS)) { // 在30秒内成功读到数据debugLogger.debug("Read resp data success");msg1 = (IMessage) readFuture.getMessage();}session.close(true);CloseFuture closeFuture = session.getCloseFuture();closeFuture.awaitUninterruptibly();if (closeFuture.isClosed())debugLogger.debug("session is closed.");return msg1;} } ----------------父类-----package com.kms.components.io;import com.kms.components.io.dto.IMessage; import com.kms.framework.core.exception.KmsException; import org.apache.mina.core.service.IoConnector;import java.net.InetSocketAddress; import java.net.SocketAddress; /*** @author: jetyou@foxmail.com* @date: 2011-7-28* @time: 14:33:41* @desc: socket瀹㈡埛绔�*/ public abstract class ClientSocket implements IClientSocket {private IoConnector connector;SocketAddress remote;SocketAddress locale;/*** @param hostIP* @param hostPort*/public ClientSocket(String hostIP, int hostPort) {remote = new InetSocketAddress(hostIP, hostPort);}/*** @param hostIP* @param hostPort* @param localIP* @param localPort*/public ClientSocket(String hostIP, int hostPort, String localIP, int localPort) {remote = new InetSocketAddress(hostIP, hostPort);locale = new InetSocketAddress(localIP, localPort);}/*** 鍙戦�淇℃伅** @param msg*/public abstract IMessage send(IMessage msg) throws KmsException;/****/public void dispose() {connector.dispose();}public SocketAddress getRemote() {return remote;}public void setRemote(SocketAddress remote) {this.remote = remote;}public SocketAddress getLocale() {return locale;}public void setLocale(SocketAddress locale) {this.locale = locale;}public IoConnector getConnector() {return connector;}public void setConnector(IoConnector connector) {this.connector = connector;} } ------父类对应的接口------ package com.kms.components.io;import com.kms.components.io.dto.IMessage; import com.kms.framework.core.exception.KmsException; public interface IClientSocket {public IMessage send(IMessage msg) throws KmsException; }解码和编码的代码的载体主要在父类里面(protocolEncoder,protocolDecoder)
package com.newyulong.upm.io.h2.codec;public class DefaultProtocolFactory extends ProtocolFactory {} ---以及子类---package com.newyulong.upm.io.h2.codec;import org.apache.mina.core.session.IoSession; import org.apache.mina.filter.codec.ProtocolCodecFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; public abstract class ProtocolFactory implements ProtocolCodecFactory,InitializingBean {private ProtocolEncoder protocolEncoder;private ProtocolDecoder protocolDecoder;public org.apache.mina.filter.codec.ProtocolEncoder getEncoder(IoSession ioSession) throws Exception {return this.protocolEncoder;}public org.apache.mina.filter.codec.ProtocolDecoder getDecoder(IoSession ioSession) throws Exception {return this.protocolDecoder;}public void setProtocolEncoder(ProtocolEncoder protocolEncoder) {this.protocolEncoder = protocolEncoder;}public void setProtocolDecoder(ProtocolDecoder protocolDecoder) {this.protocolDecoder = protocolDecoder;}public void afterPropertiesSet() throws Exception {Assert.notNull(this.protocolEncoder);Assert.notNull(this.protocolDecoder);} }
编码的类
package com.newyulong.upm.io.h2.codec;import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.session.IoSession; import org.apache.mina.filter.codec.ProtocolEncoderAdapter; import org.apache.mina.filter.codec.ProtocolEncoderOutput;import com.kms.components.io.dto.IMessage; public class ProtocolEncoder extends ProtocolEncoderAdapter {//覆盖父类的encode方法public void encode(IoSession ioSession, Object o, ProtocolEncoderOutput protocolEncoderOutput)throws Exception{IMessage message = (IMessage)o;IoBuffer buf = IoBuffer.allocate(20);buf.setAutoExpand(true);
//调用具体的编码类的encoder方法message.encoder(buf);buf.flip();System.out.println("enter in encoder ........"); // buf.asCharBuffer(); protocolEncoderOutput.write(buf);}}
具体的编码类
package com.newyulong.upm.io.h2.dto.req.XinJiang;import java.io.UnsupportedEncodingException;import org.apache.mina.core.buffer.IoBuffer;import com.kms.components.io.dto.IMessage; import com.kms.framework.core.logger.ILogger; import com.kms.framework.core.logger.LoggerFactory; import com.newyulong.upm.io.h2.util.XinJiangConstants;public class RequestInfo implements IMessage {private static final long serialVersionUID = 1L;protected ILogger operationLogger = LoggerFactory.getOperationLogger(getClass());private String packageBody;private String phoneNo; //20位业务号码private String serviceId; //12位服务编码private String serialId; //业务流水号private String bussinessUserCode; // private String bussinessAddressCode; // public void encoder(IoBuffer buf) {StringBuffer str = new StringBuffer();str.append(getPackageHeader());str.append(this.getPackageBody()); // 包体内容由请求包字段组成，每字段之间用“TAB键0x09”分隔str.append(XinJiangConstants.PACKAGE_END); // 包尾try {buf.put(str.toString().getBytes("UTF-8")); // 可以不加编码，看具体情况了.... } catch (UnsupportedEncodingException e) {e.printStackTrace();}System.out.println("upm request ====>"+ str);}public StringBuffer getPackageHeader(){operationLogger.info("营业员代码===>" + this.getBussinessUserCode()+ " 营业点代码===>" + this.getBussinessAddressCode());StringBuffer str = new StringBuffer();str.append(XinJiangConstants.VERSION); //2位协议版本号 默认值 String msglenth = (XinJiangConstants.HEAD_LENGTH + 1 + this.getPackageBody().length()) + "";str.append(msglenth); //5位数据包长度if (msglenth.length() < 5) { // 数据左对齐，不足补空格for (int i = msglenth.length(); i < 5; i++) {str.append(" ");}}str.append(checkLength(this.getSerialId()+"",20)); //20位业务流水号，交易唯一标识。客户产生传送到综合营帐数据校验包可无流水号str.append(checkLength(XinJiangConstants.RESULT_FLAG,1)); //1位操作结果标志，仅适用于响应包。1成功，0失败str.append(this.getServiceId()); //12位服务编码 str.append(checkLength(this.getPhoneNo(),20)); //20位业务号码，左对齐，不足补空格。可以是移动电话号码、IP Phone帐号、市话电//话号码、193长话帐号、寻呼号、165帐号等等。str.append(checkLength(XinJiangConstants.OPERATE_NO_TYPE,1)); //1位业务号码类型，1－电话号码，2－帐号，3－其他 str.append(checkLength(this.getBussinessAddressCode(),6)); //6位营业点代码 str.append(checkLength(this.getBussinessUserCode(),8)); //8位营业员代码String packageNo = XinJiangConstants.PACKAGE_NO;while (packageNo.length()<5) {packageNo = 0 + packageNo; }str.append(packageNo); //5位包序号，标志该包是该笔流水的第几个数据包，右对齐，左补零 str.append(checkLength(XinJiangConstants.END_PACKAGE_FLAG,1)); //1位最后一包标志，在进行多包发送的情况下，该标志用以标明最后一个数据包。//1－最后一个数据包，0－还有后续包，或者表示连接错误，I/O错误等等。str.append(checkLength(XinJiangConstants.ERROR_NO,5)); 5位错误码，包括系统操作错误和业务处理错误，由综合营帐提供。在标志为失败时System.out.println("upm request Header====>"+ str);return str;}/*** 根据定义的字段长度补空格*/public String checkLength(String str, int length) {for (int i = str.length(); i < length; i++) {str += " ";}return str;}public void decoder(IoBuffer buf) {}public String getPackageBody() {return packageBody;}public void setPackageBody(String packageBody) {this.packageBody = packageBody;}public String getPhoneNo() {return phoneNo;}public void setPhoneNo(String phoneNo) {this.phoneNo = phoneNo;}public String getServiceId() {return serviceId;}public void setServiceId(String serviceId) {this.serviceId = serviceId;}public String getSerialId() {return serialId;}public void setSerialId(String serialId) {this.serialId = serialId;}public String getBussinessAddressCode() {return bussinessAddressCode;}public void setBussinessAddressCode(String bussinessAddressCode) {this.bussinessAddressCode = bussinessAddressCode;}public String getBussinessUserCode() {return bussinessUserCode;}public void setBussinessUserCode(String bussinessUserCode) {this.bussinessUserCode = bussinessUserCode;} }
具体解码的代码此时直接覆盖了父类的doDecode方法,在此对象中注入了自己的对象protocolDecoderFactory
package com.newyulong.upm.io.h2.codec;import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.session.IoSession; import org.apache.mina.filter.codec.CumulativeProtocolDecoder; import org.apache.mina.filter.codec.ProtocolDecoderOutput;import com.kms.components.io.codec.IProtocolDecoderFactory; import com.kms.components.io.dto.IMessage; public class ProtocolDecoder extends CumulativeProtocolDecoder {private IProtocolDecoderFactory protocolDecoderFactory;public void setProtocolDecoderFactory(IProtocolDecoderFactory protocolDecoderFactory) {this.protocolDecoderFactory = protocolDecoderFactory;}protected boolean doDecode(IoSession session, IoBuffer in,ProtocolDecoderOutput out) throws Exception {//ShanXiProtocolFactory就是protocolDecoderFactory对象之一,多以protocolDecoderFactory对象调用的就是
//ShanXiProtocolFactory对象方法中的decode方法IMessage message = protocolDecoderFactory.decoder(in);if(message == null){return false;}out.write(message);return true;}}
--具体的解码对象的类此类的作用是运用配置的不同读取不同据点下的解码类工厂类
package com.newyulong.upm.io.h2;
import java.util.Map;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.util.Assert;
import com.kms.components.io.codec.IProtocolDecoderFactory;
import com.newyulong.upm.handler.IHandler;
public class ProtocolDecodeFactoryBean implements FactoryBean {
private String region; //局点名称
private Map<String, IProtocolDecoderFactory> protocolDecoderFactorys;//具体的类名配置
private Map<String, String> regionProtocolDecoderFactorys;//局点名称对应的类名
public Object getObject() throws Exception {
String tranString = regionProtocolDecoderFactorys.get(region);
IProtocolDecoderFactory protocolDecoderFactory = protocolDecoderFactorys.get(tranString);
Assert.notNull(protocolDecoderFactory);
return protocolDecoderFactory;
}
public Class getObjectType() {
return IHandler.class;
}
public void setRegion(String region) {
this.region = region;
}
public void setHandlers(Map<String, IProtocolDecoderFactory> protocolDecoderFactorys) {
this.protocolDecoderFactorys = protocolDecoderFactorys;
}
public void setRegionHandlers(Map<String, String> regionProtocolDecoderFactorys) {
this.regionProtocolDecoderFactorys = regionProtocolDecoderFactorys;
}
public boolean isSingleton() {
return false;
}
}
--具体某一个的解码类----
package com.newyulong.upm.io.h2.codec;
import org.apache.mina.core.buffer.IoBuffer;
import com.kms.components.io.codec.IProtocolDecoderFactory;
import com.kms.components.io.dto.IMessage;
import com.kms.framework.core.logger.ILogger;
import com.kms.framework.core.logger.LoggerFactory;
import com.newyulong.upm.io.h2.dto.resp.ShanXi.AccountResponse;
import com.newyulong.upm.io.h2.dto.resp.XinJiang.OpenResponse;
import com.newyulong.upm.io.h2.dto.resp.XinJiang.OrderRespone;
import com.newyulong.upm.io.h2.dto.resp.XinJiang.OrdersResponse;
import com.newyulong.upm.io.h2.util.ShanXiConstants;
public class ShanXiProtocolFactory implements IProtocolDecoderFactory {
protected ILogger debugLogger = LoggerFactory.getOperationLogger(getClass());
public IMessage decoder(IoBuffer in) {
int start = in.position();
byte previous = 0;
while (in.hasRemaining()) {
byte current = in.get();
if (current == (byte) 0x1a) {
int position = in.position();
int limit = in.limit();
try {
in.position(start);
in.limit(position);
byte[] data = new byte[limit];
in.get(data);
IoBuffer body = IoBuffer.wrap(data);
IMessage msg = parseData(body);
return msg;
} finally {
in.position(position);
in.limit(limit);
}
}
previous = current;
}
in.position(start);
return null;
}
public IMessage parseData(IoBuffer buffer) {
if (buffer.limit() < 87) {
return null;
}
// buffer.setAutoExpand(true);
buffer.position(2);
byte[] packageLength = new byte[5];
buffer.get(packageLength);
buffer.position(28);
byte[] type = new byte[12];
buffer.get(type);
buffer.position(0);
String returnType = new String(type);
byte[] data = new byte[buffer.limit()];
buffer.get(data);
debugLogger.info("Response basic data:", new String(data));
IMessage msg = null;
debugLogger.debug("业务编码===>" + returnType);
if (returnType.equals(ShanXiConstants.NUMBER_AUTHENTICATION)) //用户查询
msg = new AccountResponse();
else if (returnType.equals(ShanXiConstants.ADDED_SERVICE_APPLICANCE)||
returnType.equals(ShanXiConstants.ADDITION_PACKAGE_CHANGEMENT_APPLICANCE)||
returnType.equals(ShanXiConstants.V_ADDED_BUSSINESS_ORDER)||
returnType.equals(ShanXiConstants.SUPERPOTION_PACKAGE_ORDER) ) //订购
msg = new OpenResponse();
else if (returnType.equals(ShanXiConstants.ADDED_SERVICE_OPENUP_INQURY)) { //订购查询
msg = new OrdersResponse();
}else if (returnType.equals(ShanXiConstants.USER_GPRS_INQUIRY)||
returnType.equals(ShanXiConstants.ADDITION_PACKAGE_INQUIRY)||
returnType.equals(ShanXiConstants.V_ADDED_BUSSINESS_INQUIRY)) {//地市查询
msg = new OrderRespone();
} else if (returnType.equals(ShanXiConstants.CITY_INQUIRY)) {//地市查询
msg = null;
}else {
debugLogger.error("Response service code is not exists.",new String(data));
}
if (msg != null)
buffer.position(0);
msg.decoder(buffer);// 具体消息体解码
return msg;
}
}
--客户端事件触发类--
package com.newyulong.upm.io.h2.handler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.transport.socket.SocketSessionConfig;
import com.kms.framework.core.logger.ILogger;
import com.kms.framework.core.logger.LoggerFactory;
public class ClientIoHandler extends IoHandlerAdapter {
private ILogger operationLogger = LoggerFactory.getOperationLogger(ClientIoHandler.class);
public void sessionCreated(IoSession session) throws Exception {
operationLogger.debug("create a io session");
SocketSessionConfig cfg = (SocketSessionConfig) session.getConfig();
cfg.setReceiveBufferSize(1024);
cfg.setReadBufferSize(1024);
cfg.setKeepAlive(false);
cfg.setSoLinger(0); //这个是根本解决问题的设置
}
public void sessionOpened(IoSession session) {
System.out.println("a client session to server is created....");
}
public void messageReceived(IoSession session, Object message) {
System.out.println("read message from server ........");
// session.close(true);
}
public void messageSent(IoSession session, Object message) {
operationLogger.info("Message Successfully sent out: " + " "
+ message.getClass().getName());
}
public void exceptionCaught(IoSession session, Throwable cause) {
cause.printStackTrace();
System.out.println("an exception is occured ....");
}
public void sessionClosed(IoSession session) {
}
public void sessionIdle(IoSession session, IdleStatus status)
throws Exception {
}
}
转载于:https://www.cnblogs.com/working/p/3143714.html
总结
以上是生活随笔为你收集整理的spring + mina 作为客户端解析H2协议的使用总结的全部内容，希望文章能够帮你解决所遇到的问题。
- 上一篇： std::string 用法
- 下一篇： mac svn .a文件的上传方法