^-^
TrinityCore框架_读写数据处理以及缓冲区设计
最近更新：2024-09-23   |   字数总计：1.2k   |   阅读估时：5分钟   |   阅读量：次
  1. 为什么需要缓冲区
  2. 缓冲区如何工作
    1. （业务）读缓冲区
    2. （业务）写缓冲区
    3. 总结
  3. Asio MsgBuffer Server

为什么需要缓冲区

  1. 读写接口的形式决定的（因为我们预期向内核写入的字节数和实际写入的字节数可能不一致，读也是一样可能不一致）
  2. tcp流式传输决定的
    • tcp粘包处理的需要：我们的端目标是截出一个完整的数据包，我们有可能读到一个大于一个数据包的数据，也有可能读到一个小于数据包的数据，所以需要缓冲区
  3. 生产的速度大于消费的速度
    • 例如，客户端（生产）给服务端（消费）发送数据，假设服务端数据来不及处理，那么就需要设立缓冲区
    • 再如，业务层（生产）给内核（消费）发送数据，而内核此时处理的速度跟不上，那么也需要设立缓冲区

缓冲区如何工作

（业务）读缓冲区

  • 业务层作为消费者
  • 生产者会不断往缓冲区追加数据
    • GetWritePointer
    • GetRemainingSpace
    • 读回调中更新写指针WriteCompleted
  • 消费者会不断从缓冲区头部切割数据
    • GetReadPointer
    • ReadCompleted
  • 确保读投递要一个一个进行，不能并发投递
    • 预期读取的数据大小跟实际读取的大小可能不一致
  • 确保每次投递读请求时，平移读写指针，腾挪更大的空间接受数据，并尝试扩容缓冲区

（业务）写缓冲区

  • 业务层作为生产者
  • 生产者不断往队列中添加缓冲区
    • 构建MessageBuffer
    • writeQueue_.push(std::move(buf))
  • 消费者不断从队列中将数据发送出去
    • buf = writeQueue_.front()
    • GetReadPointer
    • GetActiveSize
    • 写回调中更新读指针 ReadCompleted
  • 确保写投递要一个一个进行，不能并发投递
    • 预期发送的数据大小跟实际发送的大小可能不一致

总结

  • 不断更新读指针和写指针

02

Asio MsgBuffer Server

  • msgbuffer.h
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    #ifndef __MESSAGEBUFFER_H_
    #define __MESSAGEBUFFER_H_

    #include <vector>
    #include <stdint.h>
    #include <cstring>

    class MessageBuffer
    {
        typedef std::vector<uint8_t>::size_type size_type;

    public:
        MessageBuffer() : _wpos(0), _rpos(0), _storage()
        {
            _storage.resize(4096);
        }

        explicit MessageBuffer(std::size_t initialSize) : _wpos(0), _rpos(0), _storage()
        {
            _storage.resize(initialSize);
        }

        MessageBuffer(MessageBuffer const& right) : _wpos(right._wpos), _rpos(right._rpos), _storage(right._storage)
        {
        }

        MessageBuffer(MessageBuffer&& right) : _wpos(right._wpos), _rpos(right._rpos), _storage(right.Move()) { }

        void Reset()
        {
            _wpos = 0;
            _rpos = 0;
        }

        void Resize(size_type bytes)
        {
            _storage.resize(bytes);
        }

        uint8_t* GetBasePointer() { return _storage.data(); }

        uint8_t* GetReadPointer() { return GetBasePointer() + _rpos; }

        uint8_t* GetWritePointer() { return GetBasePointer() + _wpos; }

        void ReadCompleted(size_type bytes) { _rpos += bytes; }

        void WriteCompleted(size_type bytes) { _wpos += bytes; }

        size_type GetActiveSize() const { return _wpos - _rpos; }

        size_type GetRemainingSpace() const { return _storage.size() - _wpos; }

        size_type GetBufferSize() const { return _storage.size(); }

        // Discards inactive data
        void Normalize()
        {
            if (_rpos)
            {
                if (_rpos != _wpos)
                    memmove(GetBasePointer(), GetReadPointer(), GetActiveSize());
                _wpos -= _rpos;
                _rpos = 0;
            }
        }

        // Ensures there's "some" free space, make sure to call Normalize() before this
        void EnsureFreeSpace()
        {
            // resize buffer if it's already full
            if (GetRemainingSpace() == 0)
                _storage.resize(_storage.size() * 3 / 2);
        }

        void Write(void const* data, std::size_t size)
        {
            if (size)
            {
                memcpy(GetWritePointer(), data, size);
                WriteCompleted(size);
            }
        }

        std::vector<uint8_t>&& Move()
        {
            _wpos = 0;
            _rpos = 0;
            return std::move(_storage);
        }

        MessageBuffer& operator=(MessageBuffer const& right)
        {
            if (this != &right)
            {
                _wpos = right._wpos;
                _rpos = right._rpos;
                _storage = right._storage;
            }

            return *this;
        }

        MessageBuffer& operator=(MessageBuffer&& right)
        {
            if (this != &right)
            {
                _wpos = right._wpos;
                _rpos = right._rpos;
                _storage = right.Move();
            }

            return *this;
        }

    private:
        size_type _wpos;
        size_type _rpos;
        std::vector<uint8_t> _storage;
    };

    #endif /* __MESSAGEBUFFER_H_ */
  • asio_server.cpp
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    #include "msgbuffer.h"
    #include <boost/asio.hpp>
    #include <boost/system/error_code.hpp>
    #include <cstddef>
    #include <memory>
    #include <queue>
    #include <iostream>

    using namespace boost::asio;
    using namespace boost::asio::ip;
    using namespace std;

    class Session : public std::enable_shared_from_this<Session>{
    public:
        Session(tcp::socket socket) : socket_(std::move(socket)), closed_(false), isWriteAsync_(false)
        {

        }
        ~Session(){
            do_close();
        }
        void Start(){
            std::cout << "accept a connection: " << socket_.remote_endpoint().address().to_string() <<std::endl;
            do_read();
        }
        void Send(const uint8_t* msg, std::size_t len){
            MessageBuffer buffer(len + 4);
            buffer.Write("S:", 2);
            buffer.Write(msg, len);
            buffer.Write("\r\n", 2);
            writeQueue_.push(std::move(buffer));
            do_write();
        }
    private:
        void do_close(){
            if(!closed_){
                closed_ = true;
                std::cout << "closed a connections: " << socket_.remote_endpoint().address().to_string() <<std::endl;
                boost::system::error_code err;
                socket_.close(err); 
            }
        }
        void do_read(){
            readBuffer_.Normalize(); // 每次都尽可能将buffer里面的内容移到最前面以充分利用空间
            readBuffer_.EnsureFreeSpace(); // 检查下是否需要扩容
            auto self(shared_from_this()); // 这里是为了保证回调的时候这个对象仍然存在，用智能指针的方式保证至少有一个引用，不然调用do_write会有问题
            socket_.async_read_some(buffer(readBuffer_.GetWritePointer(), readBuffer_.GetRemainingSpace()),[this, self](boost::system::error_code err, std::size_t length){
                if(err){
                    do_close();
                    return;
                }
                // 更新写指针的位置，length是实际的大小
                readBuffer_.WriteCompleted(length);
                std::cout << "recv length: " << length << std::endl;
                uint8_t* base = readBuffer_.GetReadPointer();
                readBuffer_.ReadCompleted(length);
                Send(base, length);
                do_read();
            });
        }
        void do_write(){
             if(isWriteAsync_)
                return; // 一个发送没有成功完成不允许发下一个，保证了发送的顺序
             isWriteAsync_ = true;
             MessageBuffer &buf = writeQueue_.front();
             auto self(shared_from_this());
             socket_.async_write_some(buffer(buf.GetReadPointer(), buf.GetActiveSize()), [this, self](boost::system::error_code err, std::size_t length){
                if(err){
                    do_close();
                    return;
                }
                isWriteAsync_ = false;
                writeQueue_.front().ReadCompleted(length);
                if(!writeQueue_.front().GetActiveSize())
                    writeQueue_.pop();
                if(!writeQueue_.empty())
                    do_write();
             });
        }
        tcp::socket socket_;
        MessageBuffer readBuffer_;
        std::queue<MessageBuffer> writeQueue_;
        bool closed_;
        bool isWriteAsync_;
    };

    class Server{
    public:
        Server(io_context &io_ctx, short port) : acceptor_(io_ctx, tcp::endpoint(tcp::v4(), port))
        {
            do_accept();
        }
    private:
        void do_accept(){
            acceptor_.async_accept([this](boost::system::error_code err, tcp::socket socket){
                if(!err){
                    std::make_shared<Session>(std::move(socket))->Start();
                }
                do_accept();
            });
        }
    private:
        tcp::acceptor acceptor_;
    };

    int main()
    {
        io_context io_ctx;
        Server s(io_ctx, 8989);
        io_ctx.run();
        return 0;
    }
2024-06-13 该篇文章被 Cleofwine 归为分类: Game

上一篇：TrinityCore框架_网络模块 下一篇：游戏后端技术_boost_asio网络编程
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题