mima开发实列

最顶层父基类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()+"");

    }

巴特西

mima开发实列

最新文章

热门文章