最顶层父基类Clinet：用于记录公共内容
切供多个Clinet继承公用
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.Timer;
import java.util.TimerTask; import org.apache.commons.lang.StringUtils;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector; import com.qinghuainvest.tsmarket.codec.HCoderFactory;
import com.qinghuainvest.tsmarket.socketclient.MinaClientHanlder; /**
* 父基类
* @author huage
*
*/
public abstract class MinaBaseClient { public abstract void startSocketReq(String code);
protected String hostName ;
protected int bindPort; /**
* 创建IOSession
* @return
*/
protected IoSession createSession(){
if( StringUtils.isBlank(hostName)) return null;
try {
NioSocketConnector connector = new NioSocketConnector();
DefaultIoFilterChainBuilder chain = connector.getFilterChain();
chain.addLast("objectFilter", new ProtocolCodecFilter(new HCoderFactory(Charset.forName("UTF-8"))));
MinaClientHanlder handler = new MinaClientHanlder();
connector.setHandler(handler);
connector.getSessionConfig().setUseReadOperation(true);
ConnectFuture cf = connector.connect(new InetSocketAddress(hostName, bindPort));
cf.awaitUninterruptibly();
return cf.getSession();
} catch (Exception e) {
e.printStackTrace();
}
return null;
} /**
* 传输数据
* @param session
* @param code
*/
protected void writeMina(IoSession session,String code){
if( session == null )return;
session.write(code + "\n");
} public static void main(String[] args) {
Integer cacheTime = 1000 * 1;
Timer timer = new Timer();
timer.schedule(new TimerTask() {
public void run() {
/*StockMina1004Client ns = new StockMina1004Client("218.1.111.62", 10003);
ns.startSocketReq(EmCommunicationCode.nqxx.getCode()+"" );*/
}
}, 1000, cacheTime);
}
}
子类继承
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; public class StockMinaClient extends MinaBaseClient{
private final Logger log = LoggerFactory.getLogger(StockMinaClient.class);
private boolean isStatus = true;
private IoSession session; public StockMinaClient(String hostName,int bindPort){
super.hostName = hostName;
super.bindPort = bindPort;
} public void startSocketReq(String code) {
if (isStatus) {
isStatus = false;
log.info("socket request start....hostName="+hostName+";bindPort="+bindPort+";requestparam="+code);
if (session == null || !session.isConnected()) {
session = createSession();
}
isStatus = true;
super.writeMina(session, code);
}
} }
子类重写(只是表明可以区别创建的对象)
import java.net.InetSocketAddress;
import java.nio.charset.Charset; import org.apache.commons.lang.StringUtils;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector; import com.qinghuainvest.tsmarket.codec.HCoderFactory;
import com.qinghuainvest.tsmarket.socketclient.MinaClient1004Hanlder; public class StockMina1004Client extends StockMinaClient{
public StockMina1004Client(String hostName, int bindPort) {
super(hostName, bindPort);
} protected IoSession createSession(){
if( StringUtils.isBlank(hostName)) return null;
try {
NioSocketConnector connector = new NioSocketConnector();
DefaultIoFilterChainBuilder chain = connector.getFilterChain();
chain.addLast("objectFilter", new ProtocolCodecFilter(new HCoderFactory(Charset.forName("UTF-8"))));
MinaClient1004Hanlder handler = new MinaClient1004Hanlder();
connector.setHandler(handler);
connector.getSessionConfig().setUseReadOperation(true);
ConnectFuture cf = connector.connect(new InetSocketAddress(hostName, bindPort));
cf.awaitUninterruptibly();
return cf.getSession();
} catch (Exception e) {
e.printStackTrace();
}
return null;
} }
需要的handler(根据实际业务调整messageReceived方法中的内容既可)
import java.util.Set; import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; public class MinaClientHanlder extends IoHandlerAdapter {
private final Logger log = LoggerFactory.getLogger(MinaClientHanlder.class);
@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
cause.printStackTrace();
} @Override
public void messageReceived(IoSession session, Object message)throws Exception {
log.info("收到行情服务器回调传送数据...");
if (message != null) {
try {
callback((String) message,session);
clear(session);
} catch (Exception e) {
log.error("message Received ", e);
//e.printStackTrace();
session.write("system error.");
}
}
} public void callback(String msg,IoSession session){
if (!"null".equals(msg)) {
//处理数据
}
} /**
* 清除session中的attribute
* 解决mina通讯中内存溢出异常
* 测试方法，未找到明确依据
* @param session
*/
private void clear(IoSession session) {
Set<Object> set = session.getAttributeKeys();
if(set==null || set.size()==0) return;
for (Object object : set) {
if(session.containsAttribute(object)) {
session.removeAttribute(object);
}
}
}
@Override
public void sessionCreated(IoSession session) throws Exception {
log.info("session Created--");
//System.out.println("session Created");
}
@Override
public void messageSent(org.apache.mina.core.session.IoSession session, java.lang.Object message) {
log.info("message Sented");
//System.out.println("message Sented--");
}
}
根据业务需要的另一个handler
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MinaClient1004Hanlder extends MinaClientHanlder {
private final Logger log = LoggerFactory.getLogger(MinaClient1004Hanlder.class);public void callback(String msg,IoSession session) {
if (!"null".equals(msg)) {
//处理数据
}
startCheck();
} }
特殊业务处理的工具类(根据业务自行处理)
package com.qinghuainvest.tsmarket.codec; import java.nio.charset.Charset; import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolEncoder; public class HCoderFactory implements ProtocolCodecFactory { private final HDecoder decoder;
private final HEncoder encoder;
// private final TextLineEncoder encoder; public HCoderFactory() {
this(Charset.defaultCharset());
} public HCoderFactory(Charset charset) {
decoder = new HDecoder();
encoder = new HEncoder();
// encoder = new TextLineEncoder();
} public ProtocolEncoder getEncoder(IoSession session) throws Exception {
return encoder;
} public ProtocolDecoder getDecoder(IoSession session) throws Exception {
return decoder;
}
}
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; public class HDecoder extends CumulativeProtocolDecoder { @Override
protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
// Remember the initial position.
int start = in.position();
// byte previous = 0;
byte[] requestMsgArray;
while (in.hasRemaining()) {
byte current = in.get();
if (current == '\n') {
// Remember the current position and limit.
int position = in.position();
int limit = in.limit();
try {
int dataLength = position - start - 1;//忽略掉\n,所以减1
in.position(start);
in.limit(position);
requestMsgArray = new byte[dataLength];
// The bytes between in.position() and in.limit()
// now contain a full CRLF terminated line.
in.get(requestMsgArray);
out.write(new String(requestMsgArray));
} finally {
// Set the position to point right after the
// detected line and set the limit to the old
// one.
in.position(position);
in.limit(limit);
}
// Decoded one line; CumulativeProtocolDecoder will
// call me again until I return false. So just
// return true until there are no more lines in the
// buffer.
// previous = current;
return true;
}
}
in.position(start); return false; }
}
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; public class HEncoder extends ProtocolEncoderAdapter { public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
String msg = (String) message;
byte[] msgArray = msg.getBytes();
IoBuffer buffer = IoBuffer.allocate(msgArray.length , false);
buffer.put(msgArray);
buffer.flip();
out.write(buffer);
out.flush();
}
}
处理调用启动写main方法调用
public void start(){
StockMinaClient client = null;
if( client == null ){
client = new StockMinaClient(market.getHostName(), market.getBindPort());
}
client.startSocketReq(EmCommunicationCode.nqxx.getCode()+"");
}