亲宝软件园·资讯

展开

C++ boost asio

tutu-hu 人气:0

一.功能介绍

  基于boost asio实现server端通信,采用one by one的同步处理方式,并且设置连接等待超时。下面给出了string和byte两种数据类型的通信方式,可覆盖基本通信场景需求。

二.string类型数据交互

  规定server与client双方交互的数据格式是string,并且server采用read_until的方式接收来自client的消息,通过delimiter(分隔符)来判断一帧数据接收完成,当未收到来自client的delimiter,那么server会一直等待,直到收到delimiter或超时。此处设置了本次回话的连接超时时间SESSION_TIMEOUT,程序中定义为2min,每次收到数据后重新计时,若是连续2min中内有收到来自client的任何消息,那么server会自动断开本次连接,并且析构本次的session。

  通过string发送和接收的数据采用ASCII码的编码方式,因此不能直接发送byte数据,不然会产生乱码(第三部分为byte数据交互);采用string数据传输方式可以方便的进行序列化与反序列化,例如采用json对象的传输的方式,可以方便的组织交互协议。

  下面是功能实现的完整程序,程序编译的前提是已经安装了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函数传入
    }
    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数据必须以"}\n"结尾
    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请求放在队列中,循环逐个处理,处理完继续阻塞
            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,调用run()函数进入io事件循环
        }
    }
}  // 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();                //在构造函数中调用do_read()函数完成对software数据的读取
        async_deadline_wait();    //set on-request-deadline
    }
    session::~session()
    {
        std::cout<<"session destruct!" << std::endl;
    }
    /**
     * @brief   从software异步读取数据并存放在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,重新执行读取操作
            }
            // 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); // 关闭socket
                socket_.close();                                    // 关闭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);            // 消息处理句柄函数
        deadline_.expires_after(SESSION_TIMEOUT);   // close session if no request,超时2min自动关闭session
    }
    /**
     * @brief   Async wait for the deadline,计时等待函数
     * @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 左值引用上,但是左值不能绑定到右值引用上)
     * 右值引用也可以看作起名,只是它起名的对象是一个将亡值。然后延续这个将亡值的生命,直到这个引用销毁的右值的生命也结束了。
     */
    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

分析一下系统执行流程:

  socket_.shutdown(asio::socket_base::shutdown_both); // 关闭socket

  socket_.close(); // 关闭socket

  deadline_.cancel(); // deadline wait计时取消

2.2 编译&&执行

编译:g++ main.cpp software.cpp -o iotest -lpthread -lboost_system -std=c++17

执行:./iotest 11112 (监听端口为11112)

2.3 程序执行结果

可以看出,client发送的每条消息都要以"}"结束,这是设定的delimter分隔符。

可以看出,当超过2min没有收到来自clinet的消息,server会自动断开连接。

  tips:client1和clinet2可同时与server建立连接并发送数据,但是server会按照连接建立的先后顺序对client发送的请求进行one by one处理,比如clinet1先与server建立了连接,那么只有等到clinet1的所有请求执行完成才会处理client2发送的请求;在等待期间client2发送的请求不会处理,但不会丢失。

三.byte类型数据交互

  上述给出了string类型数据的交互,但是string类型的数据只能采用ASCII码的方式传输,在某些场景中,例如传感器,需要交互byte类型的数据。因此下面给出了byte[hex]类型数据的交互。与上述的string数据交互流程基本一致,几点区别在下面阐述:

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数据必须以"}\n"结尾
    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请求放在队列中,循环逐个处理,处理完继续阻塞
            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,调用run()函数进入io事件循环
        }
    }
}  // 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();                //在构造函数中调用do_read()函数完成对software数据的读取
        async_deadline_wait();    //set on-request-deadline
    }
    session::~session()
    {
        std::cout<<"session destruct!" << std::endl;
    }
    /**
     * @brief   从software异步读取数据并存放在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,重新执行读取操作
            }
            // 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); // 关闭socket
                socket_.close();                                    // 关闭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自动关闭session
    }
    /**
     * @brief   Async wait for the deadline,计时等待函数
     * @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

3.2 编译&&执行

编译:g++ main.cpp software.cpp -o iotest -lpthread -lboost_system -std=c++17

执行:./iotest 11112 (监听端口为11112)

3.3 程序执行结果

加载全部内容

相关教程
猜你喜欢
用户评论