C++使用boost::asio读取LAN端口持续数据流并校验包头包尾发送数据包

现有读取文件代码 #

void sendfile(string pathname)
{
  ifstream ifs(pathname, std::ifstream::in);

  if(ifs.fail()) 
  {
      throw "error opening";
  }

  const size_t BUFFER_SIZE = 1024;
  char* buffer;
  buffer = new char[BUFFER_SIZE];

  // get file size
  ifs.seekg(0, ios::end);
  int file_size = ifs.tellg();
  ifs.seekg(0, ios::beg);

  cout <<"File size : "<<file_size<< "bytes"<<endl;

  unsigned char *data = new unsigned char [file_size];

  ifs.read((char*)data, file_size);

  for(int i=0; i < file_size; i += 2)
  {
    if (data[i] == 0xAF 
    && data[i+1] == 0xDB 
    && data[i+2] == 0xAA 
    && data[i+3] == 0x00)
    {
        if (data[i + BUFFER_SIZE - 4] == 0xFF
        && data[i + BUFFER_SIZE - 3] == 0xAA 
        && data[i + BUFFER_SIZE - 2] == 0xDF
        && data[i + BUFFER_SIZE - 1] == 0x00)
        {
            cout << "Packet" << endl;
        }
    }
  }


  delete[] data;
  ifs.close();
}

功能需求 #

• 实现逐字节读取文件或串口的不间断数据流
• 校验包头（HEADER：0xAF 0xDB 0xAA 0x00）和包尾（FOOTER：0xFF 0xAA 0xDF 0x00）
• 校验通过后将1024字节的数据包通过TCP/IP端口发送到服务端
• 采用boost::asio开发，当前已完成客户端、服务端基础代码，待实现逐字节读取未知长度不间断数据流的校验、发送逻辑
• 状态机参考：

现有已实现代码 #

客户端头文件代码 #

#include <iostream>
#include <fstream>
#include <bitset>
#include <boost/asio.hpp>
using namespace boost::asio;
using ip::tcp;
using std::string;
using std::cout;
using std::endl;
using byte = unsigned char;

class Client
{
public:
    Client(io_context& io_context,
        char host[10],
        unsigned int port)
        : socket_(io_context)
    {
        start(host, port);
    }
    ~Client()
    {
        stop();
    }
private:
    enum { CLIENT_DATA_BUFFER_SIZE = 1024 };
    char buffer[CLIENT_DATA_BUFFER_SIZE];
    boost::asio::streambuf receive_buffer;
    tcp::socket socket_;
    void start(char[10], unsigned int);
    void stop();
};

服务端头文件代码 #

TCPServer::TCPServer(io_context& io_service, char host[10], unsigned int port)
    : io_context_(io_service),
    acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
{
    cout << "Server is running ..." << endl;
    cout << "Server: got connection from " << host << " port " << port << endl;
    //Timer await
    timer = new deadline_timer(io_service, boost::posix_time::milliseconds(3100));
    timer->wait();
    std::cout << "Blocking wait(): " << 3 << " second-wait\n";
    start_accept();
}
//Creates a socket and initiates an asynchronous accept operation to wait for a new connection.
void TCPServer::start_accept()
{
    // socket definition
    connection = connectionHandler::create(io_context_);
    // asynchronous accept operation and wait for a new connection.
    acceptor_.async_accept
    (
        connection->socket(),
        boost::bind(&TCPServer::handle_accept,
        this, 
        connection,
        boost::asio::placeholders::error)
    );
}
//It services the client request, and then calls start_accept() to initiate the next accept operation.
void TCPServer::handle_accept(connectionHandler::pointer connection, const boost::system::error_code& err)
{
    if (!err)
    {
        connection->start();
    }
    start_accept();
}
void TCPServer::stop()
{
    connection->stop();
}

1. 状态机定义 #

首先定义四个状态枚举，对应包头匹配、数据包接收、包尾匹配三个核心阶段：

enum class ParserState {
    WAIT_HEADER, // 等待匹配包头
    RECEIVE_PAYLOAD, // 接收有效负载
    WAIT_FOOTER, // 等待匹配包尾
    SEND_PACKET // 校验完成准备发送
};

当前包头为4字节0xAF 0xDB 0xAA 0x00，包尾为4字节0xFF 0xAA 0xDF 0x00，数据包总长度1024字节，其中有效负载长度为1024 - 4(头) -4(尾) = 1016字节。

2. 逐字节读取与状态流转逻辑 #

在Client类中新增状态机相关成员变量、临时字节缓存以及读取对象：

private:
    ParserState state_ = ParserState::WAIT_HEADER;
    size_t matched_len_ = 0; // 已匹配的头/尾字节数
    size_t payload_received_ = 0; // 已接收的有效负载字节数
    std::array<byte, 1024> packet_buf_; // 完整数据包缓存
    byte temp_byte_; // 单字节读取缓存
    std::ifstream file_stream_; // 读文件用，若读串口替换为boost::asio::serial_port对象

新增逐字节处理函数，每次读取1字节后走状态流转：

void Client::process_next_byte(byte b) {
    switch(state_) {
        case ParserState::WAIT_HEADER:
            // 匹配包头
            if (b == (std::array<byte,4>{0xAF,0xDB,0xAA,0x00}[matched_len_])) {
                packet_buf_[matched_len_] = b;
                matched_len_++;
                if (matched_len_ == 4) {
                    // 包头匹配完成，进入接收负载阶段
                    state_ = ParserState::RECEIVE_PAYLOAD;
                    matched_len_ = 0;
                    payload_received_ = 0;
                }
            } else {
                // 匹配失败重置
                matched_len_ = 0;
            }
            break;
        case ParserState::RECEIVE_PAYLOAD:
            packet_buf_[4 + payload_received_] = b;
            payload_received_++;
            if (payload_received_ == 1016) {
                // 负载接收完成，进入匹配包尾阶段
                state_ = ParserState::WAIT_FOOTER;
            }
            break;
        case ParserState::WAIT_FOOTER:
            if (b == (std::array<byte,4>{0xFF,0xAA,0xDF,0x00}[matched_len_])) {
                packet_buf_[1020 + matched_len_] = b;
                matched_len_++;
                if (matched_len_ ==4) {
                    // 包尾匹配完成，准备发送
                    state_ = ParserState::SEND_PACKET;
                    matched_len_ =0;
                }
            } else {
                // 包尾匹配失败，重置回等待包头状态
                state_ = ParserState::WAIT_HEADER;
                matched_len_ =0;
            }
            break;
        case ParserState::SEND_PACKET:
            // 异步发送完整数据包到服务端
            boost::asio::async_write(socket_, boost::asio::buffer(packet_buf_),
                [this](const boost::system::error_code& err, size_t bytes_transferred) {
                    if (!err) {
                        cout << "发送数据包成功，长度：" << bytes_transferred << endl;
                    } else {
                        cout << "发送失败：" << err.message() << endl;
                    }
                    // 发送完成重置状态，继续处理下一个字节
                    state_ = ParserState::WAIT_HEADER;
                });
            break;
    }
}

3. 不间断数据流读取逻辑 #

如果是读文件，使用异步读每次读1字节，循环调用即可实现不间断读取：

void Client::start_read_file(const string& path) {
    if (!file_stream_.is_open()) {
        file_stream_.open(path, std::ios::binary);
        if (!file_stream_.is_open()) {
            throw std::runtime_error("打开文件失败");
        }
    }
    // 异步读1字节
    file_stream_.async_read_some(boost::asio::buffer(&temp_byte_, 1),
        [this](const boost::system::error_code& err, size_t bytes_read) {
            if (!err && bytes_read ==1) {
                process_next_byte(temp_byte_);
                // 继续读下一个字节
                start_read_file("");
            } else if (err == boost::asio::error::eof) {
                cout << "文件读取完成" << endl;
            } else {
                cout << "读文件错误：" << err.message() << endl;
            }
        });
}

如果是读串口，将文件流替换为boost::asio::serial_port对象，初始化波特率、校验位、停止位等参数后调用serial_port_.async_read_some即可，逻辑和读文件完全一致。

4. 现有代码优化点 #

• 服务端代码中不要使用timer->wait()阻塞主线程，会影响异步accept的响应，替换为async_wait异步等待即可
• Client类中可以将文件/串口读取逻辑和TCP发送逻辑完全解耦，状态机部分可以单独抽成工具类复用

内容的提问来源于stack exchange，提问作者osameh irandoust