C++基于boost asio實現(xiàn)sync tcp server通信流程詳解

更新時間：2022年07月25日 09:40:40 作者：tutu-hu

這篇文章主要介紹了C++基于boost asio實現(xiàn)sync tcp server通信的流程，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

一.功能介紹

基于boost asio實現(xiàn)server端通信，采用one by one的同步處理方式，并且設置連接等待超時。下面給出了string和byte兩種數(shù)據(jù)類型的通信方式，可覆蓋基本通信場景需求。

二.string類型數(shù)據(jù)交互

規(guī)定server與client雙方交互的數(shù)據(jù)格式是string，并且server采用read_until的方式接收來自client的消息，通過delimiter(分隔符)來判斷一幀數(shù)據(jù)接收完成，當未收到來自client的delimiter，那么server會一直等待，直到收到delimiter或超時。此處設置了本次回話的連接超時時間SESSION_TIMEOUT，程序中定義為2min，每次收到數(shù)據(jù)后重新計時，若是連續(xù)2min中內(nèi)有收到來自client的任何消息，那么server會自動斷開本次連接，并且析構(gòu)本次的session。

通過string發(fā)送和接收的數(shù)據(jù)采用ASCII碼的編碼方式，因此不能直接發(fā)送byte數(shù)據(jù)，不然會產(chǎn)生亂碼（第三部分為byte數(shù)據(jù)交互）；采用string數(shù)據(jù)傳輸方式可以方便的進行序列化與反序列化，例如采用json對象的傳輸?shù)姆绞?，可以方便的組織交互協(xié)議。

下面是功能實現(xiàn)的完整程序，程序編譯的前提是已經(jīng)安裝了boost庫，boost庫的安裝及使用方法在我的前述博客已有提到：boost庫安裝及使用

2.1 程序源碼

mian.cpp

#include "software.hpp"
int main(int argc, char** argv)
{
    if(2 != argc){
        std::cout<<"Usage: "<<argv[0]<< " port"<<std::endl;
        return -1;
    }
    try {
        boost::asio::io_context io;
        int port = atoi(argv[1]);     // get server port
        software::server(io, port);   // 開啟一個server,ip地址為server主機地址，port為mian函數(shù)傳入
    }
    catch (std::exception& e) {
        std::cout<<"main exception: " << e.what()<<std::endl;
    }
    return 0;
}

software.hpp

#ifndef __SOFTWARE_HPP__
#define __SOFTWARE_HPP__
#include <string>
#include <iostream>
#include <boost/asio.hpp>
namespace software {
    //! Session deadline duration
    constexpr auto SESSION_TIMEOUT = std::chrono::minutes(2);
    //! Protocol delimiter to software client；分隔符：接收來自client的string數(shù)據(jù)必須以"}\n"結(jié)尾
    static constexpr char const* delimiter = "}";  
    namespace asio = boost::asio;
    using tcp      = asio::ip::tcp;
    /**
     * @brief   Session for software
     * Inherit @class enable_shared_from_this<>
     * in order to give the lifecycle to io context,
     * it'll causes the lifecycle automatically end when connection break
     * (async operation will return when connection break)
     * @code
     * asio::io_context io;
     * session sess(io, std::move(socket));
     * io.run();
     * @endcode
     */
    class session
    {
    public:
        /* session constructor function */
        session(asio::io_context& io, tcp::socket socket);
        /* session destructor function */
        ~session();
    private:
        /*! Async read session socket */
        void do_read();
        /*! Async wait deadline */
        void async_deadline_wait();
        /*! software on message handler */
        void on_message(std::string&& message);
    private:
        tcp::socket        socket_;      // tcp socket
        std::string        recv_data_;   // recv buffer[string]
        asio::steady_timer deadline_;    // wait deadline time,expire it will disconnect auto 
    };
    /**
     * @brief  Start server to software(同步方式accept)
     * Will serve client one by one(同步方式)
     * @param[in]   io      The asio io context
     * @param[in]   port    The listen port
     */
    inline void server(asio::io_context& io, unsigned short port)
    {
        std::cout<<"sync server start, listen port: " << port << std::endl;
        tcp::acceptor acceptor(io, tcp::endpoint(tcp::v4(), port));
        // 一次處理一個連接[one by one]
        while (true) {
            using namespace std;
            // client請求放在隊列中，循環(huán)逐個處理，處理完繼續(xù)阻塞
            tcp::socket sock(io);
            acceptor.accept(sock);               // 一開始會阻塞在這，等待software client連接
            io.restart();  
            session sess(io, std::move(sock));   // io socket
            io.run();                            // run until session async operations done,調(diào)用run()函數(shù)進入io事件循環(huán)
        }
    }
}  // namespace software
#endif

software.cpp

#include "software.hpp"
using namespace std;
namespace software {
    /**
     * @brief       Session construct function
     * @param[in]   io      The io context
     * @param[in]   socket  The connected session socket
     */
    session::session(asio::io_context& io, tcp::socket socket)
        : socket_(std::move(socket))
        , deadline_(io)
    {
        std::cout<<"session created: " << socket_.remote_endpoint() <<std::endl;
        do_read();                //在構(gòu)造函數(shù)中調(diào)用do_read()函數(shù)完成對software數(shù)據(jù)的讀取
        async_deadline_wait();    //set on-request-deadline
    }
    session::~session()
    {
        std::cout<<"session destruct!" << std::endl;
    }
    /**
     * @brief   從software異步讀取數(shù)據(jù)并存放在recv_data_中
     */
    void session::do_read()
    {
        auto handler = [this](std::error_code ec, std::size_t length) {
            // recv data success, dispose the received data in [on_message] func
            if (!ec && socket_.is_open() && length != 0) {
                on_message(recv_data_.substr(0, length));
                recv_data_.erase(0, length);   // 將recv_data_擦除為0
                do_read();                     // Register async read operation again，重新執(zhí)行讀取操作
            }
            // error occured, shutdown the session
            else if (socket_.is_open()) {
                std::cout<<"client offline, close session" << std::endl;
                socket_.shutdown(asio::socket_base::shutdown_both); // 關(guān)閉socket
                socket_.close();                                    // 關(guān)閉socket
                deadline_.cancel();                                 // deadline wait計時取消
            }
        };
        std::cout<<"server waiting message..." << std::endl;
        // block here until received the delimiter
        asio::async_read_until(socket_, asio::dynamic_buffer(recv_data_), 
                               delimiter,           // 讀取終止條件(分隔符號)
                               handler);            // 消息處理句柄函數(shù)
        deadline_.expires_after(SESSION_TIMEOUT);   // close session if no request，超時2min自動關(guān)閉session
    }
    /**
     * @brief   Async wait for the deadline，計時等待函數(shù)
     * @pre     @a deadline_.expires_xxx() must called
     */
    void session::async_deadline_wait()
    {
        using namespace std::chrono;
        deadline_.async_wait( 
            //! lambda function
            [this](std::error_code) {
                if (!socket_.is_open())
                    return;
                if (deadline_.expiry() <= asio::steady_timer::clock_type::now()) {
                    std::cout<< "client no data more than <" 
                             << duration_cast<milliseconds>(SESSION_TIMEOUT).count()
                             << "> ms, shutdown" << std::endl;
                    socket_.shutdown(asio::socket_base::shutdown_both);
                    socket_.close();
                    return;
                }
                async_deadline_wait();
            }  
        );
    }
    /**
     * @brief       SOFTWARE on message handler
     * @param[in]   message The received message
     * &&表示右值引用，可以將字面常量、臨時對象等右值綁定到右值引用上（也可以綁定到const 左值引用上，但是左值不能綁定到右值引用上）
     * 右值引用也可以看作起名，只是它起名的對象是一個將亡值。然后延續(xù)這個將亡值的生命，直到這個引用銷毀的右值的生命也結(jié)束了。
     */
    void session::on_message(std::string&& message)
    {
        using namespace std;
        try {
            // print receive data
            std::cout<<"recv from client is: "<<message<<std::endl;
            // response to client
            string send_buf = "hello client, you send data is: " + message;
            asio::write(socket_, asio::buffer(send_buf));
        }
        catch (exception& ex) {
            std::cout<<"some exception occured: "<< ex.what() << std::endl;
        }
    }
}  // namespace software

分析一下系統(tǒng)執(zhí)行流程：

在main函數(shù)中傳入io和port，調(diào)用 software.hpp中的server(asio::io_context& io, unsigned short port)函數(shù)。
在server()函數(shù)中while(True)循環(huán)體中accept來自client的連接，每次接收到一個client的連接會創(chuàng)建一個session對象，在session對象中處理本次的連接socket。注意，此處采用的是one by one的同步處理方式，只有上一個session處理完成才能處理下一個session的請求，但是同步發(fā)送的請求消息不會丟失，只是暫時不會處理和返回；總的來說，server會按照請求的順序進行one by one處理。
session對象創(chuàng)建時會調(diào)用構(gòu)造函數(shù)，其構(gòu)造函數(shù)主要做了兩件事情：一是調(diào)用do_read()函數(shù)進行等待讀取來自client的數(shù)據(jù)并處理；二是通過async_deadline_wait()設置本次session連接的超時處理方法，超時時間默認設置為SESSION_TIMEOUT：deadline_.expires_after(SESSION_TIMEOUT)。
在do_read()函數(shù)中采用async_read_until()函數(shù)讀取來自client的數(shù)據(jù)，async_read_until()函數(shù)會將傳入的delimiter分隔符作為本次接收的結(jié)束標識。
當判斷本次接收數(shù)據(jù)完成后，會調(diào)用handler句柄對消息進行處理，在handler句柄中主要做了兩件事情：一是將收到的string信息傳入到on_message()消息處理函數(shù)中進行處理，只有當本條消息處理完成后才能接收下一條消息并處理，消息會阻塞等待，但是不會丟失；二是在消息處理完成后再次調(diào)用do_read()函數(shù)，進入read_until()等待消息，如此循環(huán)…
當發(fā)生錯誤或異常，在hander中會關(guān)閉本次socket連接，并且不會再調(diào)用其他循環(huán)體，表示本次session通信結(jié)束，之后調(diào)用析構(gòu)函數(shù)析構(gòu)session對象。

socket_.shutdown(asio::socket_base::shutdown_both); // 關(guān)閉socket

socket_.close(); // 關(guān)閉socket

deadline_.cancel(); // deadline wait計時取消

在on_message()消息處理函數(shù)中會對收到的string數(shù)據(jù)進行處理（上述程序中以打印代替），然后調(diào)用asio::write(socket_, asio::buffer(send_buf))將response發(fā)送給client。

2.2 編譯&&執(zhí)行

編譯：g++ main.cpp software.cpp -o iotest -lpthread -lboost_system -std=c++17

執(zhí)行：./iotest 11112 （監(jiān)聽端口為11112）

2.3 程序執(zhí)行結(jié)果

可以看出，client發(fā)送的每條消息都要以"}"結(jié)束，這是設定的delimter分隔符。

可以看出，當超過2min沒有收到來自clinet的消息，server會自動斷開連接。

tips：client1和clinet2可同時與server建立連接并發(fā)送數(shù)據(jù)，但是server會按照連接建立的先后順序?qū)lient發(fā)送的請求進行one by one處理，比如clinet1先與server建立了連接，那么只有等到clinet1的所有請求執(zhí)行完成才會處理client2發(fā)送的請求；在等待期間client2發(fā)送的請求不會處理，但不會丟失。

三.byte類型數(shù)據(jù)交互

上述給出了string類型數(shù)據(jù)的交互，但是string類型的數(shù)據(jù)只能采用ASCII碼的方式傳輸，在某些場景中，例如傳感器，需要交互byte類型的數(shù)據(jù)。因此下面給出了byte[hex]類型數(shù)據(jù)的交互。與上述的string數(shù)據(jù)交互流程基本一致，幾點區(qū)別在下面闡述：

將session類中的string recv_data_;替換成u_int8_t recv_data_[MAX_RECV_LEN];
數(shù)據(jù)讀取方式由read_until()改為：socket_.async_receive(asio::buffer(recv_data_,MAX_RECV_LEN),handler);
on_message()數(shù)據(jù)處理函數(shù)變?yōu)椋簐oid on_message(const u_int8_t* recv_buf,std::size_t recv_len)；
數(shù)據(jù)發(fā)送方式變?yōu)椋簊ocket_.async_send(asio::buffer(recv_buf,recv_len),[](error_code ec, size_t size){});

3.1 程序源碼

mian.cpp

同上

software.hpp

#ifndef __SOFTWARE_HPP__
#define __SOFTWARE_HPP__
#include <string>
#include <iostream>
#include <boost/asio.hpp>
#define MAX_RECV_LEN 2048 
namespace software {
    //! Session deadline duration
    constexpr auto SESSION_TIMEOUT = std::chrono::minutes(2);
    //! Protocol delimiter to software client；分隔符：接收來自client的string數(shù)據(jù)必須以"}\n"結(jié)尾
    static constexpr char const* delimiter = "}";  
    namespace asio = boost::asio;
    using tcp      = asio::ip::tcp;
    /**
     * @brief   Session for software
     * Inherit @class enable_shared_from_this<>
     * in order to give the lifecycle to io context,
     * it'll causes the lifecycle automatically end when connection break
     * (async operation will return when connection break)
     * @code
     * asio::io_context io;
     * session sess(io, std::move(socket));
     * io.run();
     * @endcode
     */
    class session
    {
    public:
        /* session constructor function */
        session(asio::io_context& io, tcp::socket socket);
        /* session destructor function */
        ~session();
    private:
        /*! Async read session socket */
        void do_read();
        /*! Async wait deadline */
        void async_deadline_wait();
        /*! software on message handler */
        void on_message(const u_int8_t* recv_buf,std::size_t recv_len);
    private:
        tcp::socket socket_;                 // tcp socket
        u_int8_t recv_data_[MAX_RECV_LEN];   // recv buffer[byte]
        asio::steady_timer deadline_;        // wait deadline time,expire it will disconnect auto 
    };
    /**
     * @brief  Start server to software(同步方式accept)
     * Will serve client one by one(同步方式)
     * @param[in]   io      The asio io context
     * @param[in]   port    The listen port
     */
    inline void server(asio::io_context& io, unsigned short port)
    {
        std::cout<<"sync server start, listen port: " << port << std::endl;
        tcp::acceptor acceptor(io, tcp::endpoint(tcp::v4(), port));
        // 一次處理一個連接[one by one]
        while (true) {
            using namespace std;
            // client請求放在隊列中，循環(huán)逐個處理，處理完繼續(xù)阻塞
            tcp::socket sock(io);
            acceptor.accept(sock);               // 一開始會阻塞在這，等待software client連接
            io.restart();  
            session sess(io, std::move(sock));   // io socket
            io.run();                            // run until session async operations done,調(diào)用run()函數(shù)進入io事件循環(huán)
        }
    }
}  // namespace software
#endif

software.cpp

#include "software.hpp"
using namespace std;
namespace software {
    /**
     * @brief       Session construct function
     * @param[in]   io      The io context
     * @param[in]   socket  The connected session socket
     */
    session::session(asio::io_context& io, tcp::socket socket)
        : socket_(std::move(socket))
        , deadline_(io)
    {
        std::cout<<"session created: " << socket_.remote_endpoint() <<std::endl;
        do_read();                //在構(gòu)造函數(shù)中調(diào)用do_read()函數(shù)完成對software數(shù)據(jù)的讀取
        async_deadline_wait();    //set on-request-deadline
    }
    session::~session()
    {
        std::cout<<"session destruct!" << std::endl;
    }
    /**
     * @brief   從software異步讀取數(shù)據(jù)并存放在recv_data_中
     */
    void session::do_read()
    {
        auto handler = [this](std::error_code ec, std::size_t length) {
            // recv data success, dispose the received data in [on_message] func
            if (!ec && socket_.is_open() && length != 0) {
                on_message(recv_data_, length);
                memset(recv_data_,0,sizeof(recv_data_));// 將recv_data_擦除為0
                do_read();                     // Register async read operation again，重新執(zhí)行讀取操作
            }
            // error occured, shutdown the session
            else if (socket_.is_open()) {
                std::cout<<"client offline, close session" << std::endl;
                socket_.shutdown(asio::socket_base::shutdown_both); // 關(guān)閉socket
                socket_.close();                                    // 關(guān)閉socket
                deadline_.cancel();                                 // deadline wait計時取消
            }
        };
        std::cout<<"server waiting message..." << std::endl;
        //block here to receive some byte from client
        socket_.async_receive(asio::buffer(recv_data_,MAX_RECV_LEN),handler);
        deadline_.expires_after(SESSION_TIMEOUT);   // close session if no request，超時2min自動關(guān)閉session
    }
    /**
     * @brief   Async wait for the deadline，計時等待函數(shù)
     * @pre     @a deadline_.expires_xxx() must called
     */
    void session::async_deadline_wait()
    {
        using namespace std::chrono;
        deadline_.async_wait( 
            //! lambda function
            [this](std::error_code) {
                if (!socket_.is_open())
                    return;
                if (deadline_.expiry() <= asio::steady_timer::clock_type::now()) {
                    std::cout<< "client no data more than <" 
                             << duration_cast<milliseconds>(SESSION_TIMEOUT).count()
                             << "> ms, shutdown" << std::endl;
                    socket_.shutdown(asio::socket_base::shutdown_both);
                    socket_.close();
                    return;
                }
                async_deadline_wait();
            }  
        );
    }
    /**
     * @brief       SOFTWARE on message handler
     * @param[in]   recv_buf The received byte array address
     * @param[in]   recv_len The received byte length
     */
    void session::on_message(const u_int8_t* recv_buf,std::size_t recv_len)
    {
        using namespace std;
        try {
            // print receive data
            std::cout<<"recv data length is: "<<recv_len<<"   data is: ";
            for(int i = 0; i<recv_len; i++)
                printf("%x ",recv_buf[i]);
            std::cout<<std::endl;
            // response to client
            socket_.async_send(asio::buffer(recv_buf,recv_len),[](error_code ec, size_t size){});
        }
        catch (exception& ex) {
            std::cout<<"some exception occured: "<< ex.what() << std::endl;
        }
    }
}  // namespace software