Mina 之 入门篇（1）

最近自己在工作之余做一个金融类的项目（类似股票），采用mina开源框架进行服务端和客户端之间交互。开始学学mina。转入正题...

（1）首先需要引入以下几个包：

 

mina-core-2.0.x.jar，slf4j-api-1.6.x.jar，slf4j-log4j12-1.6.x.jar

这里需要注意下：slf4j-api-1.6.x.jar，slf4j-log4j12-1.6.x.jar 必须是同样二级版本号，即1.6版本，如果一个是1.5版本、一个是1.6版本，运行时会报错：java.lang.NoSuchMethodError: org.slf4j.helpers.MessageFormatter.format，原因是包冲突。

（2）创建一个主服务类：TradeSocket

public class TradeSocket extends SocketBase implements ISocket {
.....
@Override
    public boolean initSocket(Config config) {
        logger.info("正在初始化 tradeSocket");
        TradeSocketConfig tradeConfig = (TradeSocketConfig) config;
        this.setSocket_IP(tradeConfig.getIp());
        logger.info("正在初始化 tradeSocket的Ip");
        this.setSocket_port(tradeConfig.getPost());
        logger.info("正在初始化 tradeSocket的端口");
        this.setHandlerAdapter(new TradeSocketHandler());
        logger.info("正在初始化 tradeSocket的处理器");
        this.setAcceptor(new NioSocketAcceptor());
        logger.info("正在初始化 tradeSocket的接收器");

        logger.info("成功初始化 tradeSocket");
        logger.info("tradeSocket服务IP:" + this.getSocket_IP());
        logger.info("tradeSocket服务端口号:" + this.getSocket_port());
        return true;
    }

    @Override
    public boolean startSocket() {
        boolean result = false;
        if (this.isStart() == false) {
            initSocket(this.tradeSocketConfig);
            // 设置过滤器，ObjectSerializationCodecFactory可以对象序列化方式在服务端和客户端之间传递数据
            this.getAcceptor().getFilterChain().addLast(DGConstants.SERVER_SOCKET_FILTER_CHIN, new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
             // 设置读取数据的缓冲区大小
            this.getAcceptor().getSessionConfig().setReadBufferSize(2048);
            // 设置长连接
            this.getAcceptor().getSessionConfig().setKeepAlive(true);
            // 读写通道10秒内无操作进入空闲状态
            this.getAcceptor().getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
            this.getAcceptor().setHandler(this.getHandlerAdapter());
            try {
                this.getAcceptor().bind(new InetSocketAddress(this.getSocket_IP(), this.getSocket_port()));
                this.setStart(true);
                result = true;
                logger.info("成功启动 tradeSocket");
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
        return result;
    }

    @Override
    public boolean closeSocket() {
        this.getAcceptor().dispose();
        this.setStart(false);
        logger.error("关闭了交易Socket");
        return false;
    }

 （3）创建一个handler来管理事件，是业务处理关注点

 

 

public class TradeSocketHandler extends IoHandlerAdapter {
public static Logger logger = Logger.getLogger(TradeSocketHandler.class);

    @Override
    public void sessionCreated(IoSession session) throws Exception {
        super.sessionCreated(session);
        logger.info("服务端与客户端创建连接...");
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {
        InetSocketAddress remoteAddress = (InetSocketAddress) session.getRemoteAddress();
        String clientIp = remoteAddress.getAddress().getHostAddress();
        logger.info("服务器打开Session ID=" + session.getId());
        logger.info( "服务端与客户端["+session.getRemoteAddress()+"]连接打开...");

//        ConnectResponseMessage connectResponseMessage = new ConnectResponseMessage();
//        connectResponseMessage.setConnectResponse(ConnectResponseMessage.ConnectSuccess);
//        session.write(connectResponseMessage);
                
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        super.sessionClosed(session);
        logger.info("服务端与客户端断开连接");
    }

    @Override
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
        super.sessionIdle(session, status);
        logger.info("服务端进入空闲状态...");
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
        logger.error(cause.getMessage(), cause);
        super.exceptionCaught(session, cause);
    }

    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        if (message instanceof BaseMessage) {
            BaseMessage baseMessage = (BaseMessage) message;
            logger.info("服务器收到的消息为：" + baseMessage);
            if (baseMessage.getMsgType() == MessageType.LoginMessage_Type) {
                LoginMessage loginMessage = (LoginMessage) baseMessage;
                UserConnect userConnect = new UserConnect();
                userConnect.setTradeSession(session);
                loginMessage.setUserid(loginMessage.getUserid() + session.hashCode());
                ServerCache cache = ServerCache.getInstance();
                cache.addUserConnect(loginMessage.getUserid(), userConnect);
                userConnect.getTradeSession().setAttribute("userID", loginMessage.getUserid());
                userConnect.getTradeSession().setAttribute("sessionID", session.hashCode());
                //这里主要是处理各种业务
                ServerReceivedMessageProcess.getInstance().processMessage(loginMessage);
            }
        }
    }

    @Override
    public void messageSent(IoSession session, Object message) throws Exception {
        super.messageSent(session, message);
    }

 ServerReceivedMessageProcess.getInstance().processMessage(.....);方法是根据从客户端传过来不同message，实例化出不同处理类。

（4）创建客户端主类

 

public class ClientTradeSocket extends ClientSocketBase {

    private static Logger logger = Logger.getLogger(ClientTradeSocket.class);
    private static ClientTradeSocket instance;

    public static ClientTradeSocket getInstance() {
        if (instance == null) {
            instance = new ClientTradeSocket();
        }
        return instance;
    }

    private ClientTradeSocket() {
    }

    @Override
    public boolean closedSocket() {
        this.getConnector().dispose();
        this.setStart(false);
        logger.info("客户关闭了与服务器的连接");
        return true;
    }

    @Override
    public void initSocket() {
        this.setHandlerAdapter(new ClientTradeHandler());
        this.setConnector(new NioSocketConnector());
        logger.info("客户初始化了与服务器的连接");
    }

    @Override
    public void startSocket() {
        this.initSocket();
        try {
            if (this.isStart() == false) {
                /**
                 * 创建接收数据的过滤器
                 */
                DefaultIoFilterChainBuilder chain = this.getConnector().getFilterChain();
                chain.addLast(DGConstants.SERVER_SOCKET_FILTER_CHIN, new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
                this.getConnector().setHandler(this.getHandlerAdapter());
                /**
                 * 设置连接超时的时间 为：一分钟 这个超时不可以过小
                 */
                this.getConnector().setConnectTimeoutMillis(ClientSocketBase.timeOutMillis);
                logger.info("ip/port:" + this.getServer_IP_Port() + "/" + this.getServer_Socket_Port());
                InetSocketAddress socketAddress = new InetSocketAddress(this.getServer_IP_Port(), this.getServer_Socket_Port());
                /**
                 * 建立连接
                 */
                this.connectFuture = this.getConnector().connect(socketAddress);
                //等待连接，这句话比较重要，如果没有，则后面的this.connectFuture.isConnected()为false的。
                //原因在于：connect方法是异步的，awaitUninterruptibly方式阻塞主线程，等待服务端返回，从而实现同步
                //如果不用awaitUninterruptibly，可以采用this.connectFuture.addListener方法进行回调
                 this.connectFuture.awaitUninterruptibly();
                //连接后，获得session
                if (this.connectFuture.isConnected()) {
                    logger.info("已经建立用户连接");
                    this.setStart(true);
                    logger.info("客户开启了与服务器的连接");
                    ConnectSession session = ConnectSession.getInstance();
                    //将session保存到缓存中
                    session.setTradeSession(this.connectFuture.getSession());
                } else {
                    logger.info("不能建立用户连接");
                    this.setStart(false);
                    ConnectSession session = ConnectSession.getInstance();
                    session.setTradeSession(null);
                }
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            this.setStart(false);
        }

    }
}

 （5）客户端业务处理类handler

 

 

public class ClientTradeHandler extends IoHandlerAdapter {

    private static Logger logger = Logger.getLogger(ClientTradeHandler.class);
    private String user;


    public void setUser(String user) {
        this.user = user;
    }

    public String getUser() {
        return this.user;
    }

    /**
     *
     * 当接口中其他方法抛出异常未被捕获时触发此方法
     */
    @Override
    public void exceptionCaught(IoSession ioSession, Throwable throwable) throws Exception {
    }

    /**
     * 当接收到消息后被触发
     *
     */
    @Override
    public void messageReceived(IoSession ioSession, Object msg) throws Exception {
        BaseMessage message = (BaseMessage) msg;
        logger.info("接受到得消息为：" + message);
        if(message!=null) {
            if(message instanceof ConnectResponseMessage) {
                ConnectResponseMessage connectResponseMessage = (ConnectResponseMessage)message;
                if(connectResponseMessage.getConnectResponse() == ConnectResponseMessage.ConnectSuccess) {
                     ConnectSession clientConnect = ConnectSession.getInstance();
                    clientConnect.setTradeSession(ioSession);
                }
            } else {
                //根据服务端返回的message，实例化不同处理类
                ClientReceivedMessageProcess messageProcess = new ClientReceivedMessageProcess();
                messageProcess.processMessage(message);
            }
        }
       
    }

    /**
     * 当发送消息后被触发
     *
     */
    @Override
    public void messageSent(IoSession ioSession, Object msg) throws Exception {
    }

    /**
     * 当会话关闭时被触发
     *
     */
    @Override
    public void sessionClosed(IoSession ioSession) throws Exception {
        logger.info("与服务器断开连接！！");
//        EIMTrayIcon trayIcon = EIMTrayIcon.getInStance();
//        trayIcon.showIcon(EIMClientConfig.OffLineTryIcon_Type);
    }

    /**
     * 当会话创建时被触发
     *
     */
    @Override
    public void sessionCreated(IoSession ioSession) throws Exception {
    }

    /**
     *
     * 当会话空闲时被触发
     */
    @Override
    public void sessionIdle(IoSession ioSession, IdleStatus msg) throws Exception {
    }

    /**
     *
     * 当会话开始时被触发
     */
    @Override
    public void sessionOpened(IoSession ioSession) throws Exception {
        logger.info("会话已经打开");
    }
}

 

最后，可以进行运行测试了。。。。