Mina 之入门篇（1）

心已赠人 2022-04-14 05:39 175阅读 0赞

最近自己在工作之余做一个金融类的项目（类似股票），采用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("会话已经打开");
        }
    }

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