目錄

• 服務端
• 客戶端
• 通訊
• 自定義工廠編解碼
• 解碼器
• 編碼器
• 總結
• # 加入戰隊
  • 微信公眾號

---

title: Mina服務端客戶端通訊
date: 2018-09-30 09:00:30
tags:
- [mina]
- [tcp]
categories:
- [程式設計]
permalink: zxh
---

前兩章節已經完整的介紹了理論部分，今天我們就利用這些理論來實現tcp協議的c/s 通訊。首先我們簡單回顧下之前的介紹，

在mina中我們的客戶端和服務端簡直就是一模一樣，只是我們用不同介面卡。但是他的資料處理流程是一樣的。今天我們就重點看看如何建立服務端、客戶端
並且處理兩者之間的訊息通訊處理

服務端

服務端和客戶端不同的就是我們建立的監聽物件不同而已，客戶端傳送訊息到服務端，服務端需要經歷過濾器的處理才能到達訊息中心，但是在過濾器中我們就需要將訊息進行解碼，然後才會到訊息接收的地方處理我們的業務。正常情況下我們處理完訊息需要對客戶端進行迴應。迴應的時候也會經歷過濾器中的編碼邏輯，進行資料編碼然後傳送。資訊傳送到客戶端我們可以看成服務端的方向。也是需要進行編解碼的。下面看看服務端的建立程式碼

//建立監聽物件
IoAcceptor acceptor = new NioSocketAcceptor();
TextLineCodecFactory textLineCodecFactory =
        new TextLineCodecFactory(Charset.forName("utf-8"), LineDelimiter.WINDOWS.getValue(),
                LineDelimiter.WINDOWS.getValue());
//新增過濾器
acceptor.getFilterChain().addLast("logger",new LoggingFilter());
acceptor.getFilterChain().addLast("protocal",new ProtocolCodecFilter(
        textLineCodecFactory
));
//設定時間處理的handler
acceptor.setHandler(new ServerMessageHandler());
//設定讀取資料快取區的大小
acceptor.getSessionConfig().setReadBufferSize(Constaint.READSIZE);
//設定多久沒有訊息就進入空閒狀態
acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE,Constaint.IDLETIME);
//繫結埠
try {
    acceptor.bind(new InetSocketAddress(Constaint.REMOTE_PORT));
} catch (IOException e) {
    logger.error(String.format("bind %s error",Constaint.REMOTE_PORT));
    e.printStackTrace();
}
logger.info(String.format("bind %s success",Constaint.REMOTE_PORT));
 

客戶端

    
//建立監聽物件
IoConnector connector = new NioSocketConnector();
TextLineCodecFactory textLineCodecFactory =
        new TextLineCodecFactory(Charset.forName("utf-8"), LineDelimiter.WINDOWS.getValue(),
                LineDelimiter.WINDOWS.getValue());
//新增過濾器
//日誌過濾器 。  sltf日誌設定
connector.getFilterChain().addLast("logger",new LoggingFilter());
//在這個過濾器中提供了編解碼，這裡的編碼是以資訊中已\r\n結尾算是一條資訊 
connector.getFilterChain().addLast("protocal",new ProtocolCodecFilter(
        new SocketFactory()
));
//設定時間處理的handler ， 提供session生命週期的監聽函式，訊息接受，傳送的函式
connector.setHandler(new ClientMessageHandler());
//設定讀取資料快取區的大小
connector.getSessionConfig().setReadBufferSize(Constaint.READSIZE);
//設定多久沒有訊息就進入空閒狀態
connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE,Constaint.IDLETIME);
ConnectFuture future = connector.connect(new InetSocketAddress(Constaint.REMOTE_IP,Constaint.REMOTE_PORT));
//是非同步處理，這裡不會造成阻塞
future.addListener(new IoFutureListener<IoFuture>() {
    @Override
    public void operationComplete(IoFuture ioFuture) {
        logger.info("連線準備完成");
        IoSession session = ioFuture.getSession();

    }
});
 

通訊

• 其實上面服務端，客戶端兩邊建立好就應經在通訊了，在上面建立的時候我們發現。建立的時候需要指定訊息處理器(IoHandlerAdapter) , 這個在IoService中會排在IoFilter之後執行。在過濾器執行之後我們就會呼叫我們的訊息處理器。

    private static Logger logger = LogManager.getLogger(ServerMessageHandler.class);
    public void sessionCreated(IoSession session) throws Exception {
        super.sessionCreated(session);
        logger.info("sessionCreated");
    }

    public void sessionOpened(IoSession session) throws Exception {
        super.sessionOpened(session);
        try {
            IoBuffer buffer = IoBuffer.allocate(30);
            buffer.clear();
            buffer.putString("quit\r\n", Charset.forName("utf-8").newEncoder());
            buffer.flip();
            session.write(buffer);
        } catch (Exception e) {
            logger.error(e.toString());
        }
        logger.info("sessionOpened");
    }

    public void sessionClosed(IoSession session) throws Exception {
        super.sessionClosed(session);
        logger.info("sessionClosed");
    }

    public void sessionIdle(IoSession session, IdleStatus idleStatus) throws Exception {
        super.sessionIdle(session,idleStatus);
        try {
            IoBuffer buffer = IoBuffer.allocate(30);
            buffer.clear();
            buffer.putString("quit\r\n", Charset.forName("utf-8").newEncoder());
            buffer.flip();
            session.write(buffer);
        } catch (Exception e) {
            logger.error(e.toString());
        }
        //        logger.info("sessionIdle");
    }

    public void exceptionCaught(IoSession ioSession, Throwable throwable) throws Exception {
        logger.info("exceptionCaught");
        throwable.printStackTrace();
    }

    public void messageReceived(IoSession session, Object message) throws Exception {
        super.messageReceived(session, message);
        String info = message.toString();
        Date date = new Date(System.currentTimeMillis());
        SimpleDateFormat sdf = new  SimpleDateFormat("yy-MM-dd HH:mm:ss");
        String time = sdf.format(date);
        session.write(time);
        System.out.println("接收到的訊息："+info);
    }

    public void messageSent(IoSession session, Object message) throws Exception {
        super.messageSent(session, message);
        logger.info("messageSent");
    }

• 這裡訊息處理器，提供了幾個時刻可以控制，比如session建立、銷燬的時候執行的地方。訊息接收的地方，訊息傳送成功的地方。這些控制力度可以根據我們的需要進行適度的複寫。

自定義工廠編解碼

• 工廠是提供編解碼的方法。這個工廠是載入在ProtocolCodecFilter這個過濾器中的。我們也可以自定義過濾器，在自定義的過濾器中我們也可以載入我們自定義的工廠，實現編解碼。我們在編解碼的地方，就可以加入我們的業務程式碼。比如解碼通過約定的協議方式讀取到內容後通過ProtocolDecoderOutput 將訊息寫出去就可以在我們的IoHandlerAdapter的messageReceived方法中獲取到訊息。然後業務書寫。這樣做到程式碼的解耦。

public class SocketFactory  implements ProtocolCodecFactory {
    private MessageDecoder decoder;
    private MessageEncoder encoder;

    public SocketFactory() {
        decoder = new MessageDecoder();
        encoder = new MessageEncoder();
    }

    public ProtocolDecoder getDecoder(IoSession session) throws Exception {
        return this.decoder;
    }

    public ProtocolEncoder getEncoder(IoSession session) throws Exception {
        return this.encoder;
    }
}

解碼器

• 上面的工廠就是提供編解碼的。和我們生活中一樣工廠提供功能，但是實際並不是工廠做的，工廠可能只代理功能，僅僅是個加工廠而已。mina通訊也是如此。真正編解碼的並不是工廠執行的，本節將揭露解碼者CumulativeProtocolDecoder

• 解碼器寫好之後只需要在上面自定義工廠中建立就好了。至於自定義編碼器只需要繼承CumulativeProtocolDecoder這個類就好了。而且複寫doDecode方法就好了。這個方法的返回值是boolean型別。返回值不同代表意義不一。這裡需要重點理清楚

+ true:  返回true表示你已經從CumulativeProtocolDecoder的訊息中消費了資訊，在編碼器中返回true之前也應該呼叫ProtocolDecoderOutput 的wirte將訊息釋出到IoHandAdaptor中進行業務處理。但是這裡會出現其他情況，應為我們服務端客戶端是長連線所以在我們訊息中訊息是不斷髮過來的，我們快取中的訊息可能是完整一條訊息，也可能不夠一整條訊息，也可能是一整條多了一點，
   1、如果不是一條完整（半包）的那麼我們返回falsed等待客戶端繼續傳送
   2、如果正好是一整條，那麼我們接受到之後返回true的時候我們快取中就沒有資料了，在CumulativeProtocolDecoder會停止對解碼中doDecode的呼叫了，這種情況不會出現意外
   3、資料比一條完整資訊(粘包)多，那麼我們處理到一條資訊後也需要返回true,但是CumulativeProtocolDecoder會將剩餘的快取繼續拼裝，剩餘訊息就相當於內部進行了第二次解碼。如果不過那麼相當於上面第一種情況

   記住三種情況  半包 、 正常 、 粘包
+ false: 返回false就是快取中的資料不夠我們一整條訊息，需要繼續等待客戶端的訊息。CumulativeProtocolDecoder中的快取機制會不斷的將客戶端發過來的資料拼接到快取中

public class MessageDecoder extends CumulativeProtocolDecoder {
    /**
     * 此方法return true : 表示父類中CumulativeProtocolDecoder會不斷的呼叫此方法進行訊息的消費
     *       return false: 表示訊息已經消費完全了，快取中就算有資料也不會再消費了。等待再次客戶端
     *       傳送訊息時會觸發訊息傳送介面，此時會將新舊訊息拼接再一起進行處理
     * @param ioSession
     * @param ioBuffer
     * @param protocolDecoderOutput
     * @return
     * @throws Exception
     */
    @Override
    protected boolean doDecode(IoSession ioSession, IoBuffer ioBuffer, ProtocolDecoderOutput protocolDecoderOutput) throws Exception {
        IoBuffer buffer = IoBuffer.allocate(10);
        while (ioBuffer.hasRemaining()) {
            if (ioBuffer.remaining()<3) {
                //繼續接受
                return false;
            }
            //獲取三個位元組
            int oldLimit = ioBuffer.limit();
            ioBuffer.limit(ioBuffer.position()+3);
            String text = ioBuffer.getString(Charset.forName("UTF-8").newDecoder());
            protocolDecoderOutput.write(text);
            ioBuffer.limit(oldLimit);
            if (ioBuffer.hasRemaining()) {
                return true;
            }
        }
        return false;
    }
}

編碼器

• 編碼器相對解碼器簡單很多，編碼器就是加入我們的協議，正常情況就是我們業務程式碼中訊息是一個Java實體，我們需要做的是將Java實體按照協議轉換成IoBuffer進行傳送。但是我們mina中傳送訊息是通過IoSession中write方法傳送的。我們檢視原始碼發現在IoSession.write(Object o)，傳送的如果是IoBuffer那麼就不經過我們的編碼器，否則會經過我們編碼器進行編碼最終將轉換後的IoBuffer傳送出去。

public class MessageEncoder extends ProtocolEncoderAdapter {
    @Override
    public void encode(IoSession ioSession, Object o, ProtocolEncoderOutput protocolEncoderOutput) throws Exception {
        //TODO  根據協議編碼
        //組裝好之後  ioSession.write(IoBuffer)寫出
        System.out.println(o);
    }
}

總結

加入戰隊

# 加入戰隊

微信公眾號

相關推薦