实现一个异步 WebSocket 服务器（AsyncWebSocketServer），使其与之前定义的 INetwork 接口兼容，并能够处理 WebSocket 协议的握手、消息收发和连接管理。

以下是基于 Boost.Beast 和 Boost.Asio 的完整实现，结合之前提到的 main 函数（修复了 lambda 捕获问题），并详细解释 WebSocket 协议的实现细节。

代码将在 Visual Studio 2022 中运行，支持 TCP、UDP 和 WebSocket 协议的统一接口。

涵盖实现步骤、代码示例、测试方法和注意事项。

一、目标与背景

  • 目标：
    • 实现 AsyncWebSocketServer 类，继承 INetwork 接口，支持异步 WebSocket 通信。
    • 确保与 main 函数兼容，处理 lambda 捕获问题（network->Send）。
    • 支持多客户端连接、异步消息收发和错误处理。
  • 背景：
    • main 函数使用 std::unique_ptr<INetwork> 管理服务器对象，通过协议字符串（"WEBSOCKET"）选择实现。
    • 之前的问题（lambda 未捕获 network）已通过添加 [&network] 解决。
    • WebSocket 协议基于 TCP，通过 HTTP 握手建立连接，使用数据帧传输消息。

二、实现 AsyncWebSocketServer

1. 前提条件

  • 依赖库：
    • Boost.Beast（处理 WebSocket 协议）。
    • Boost.Asio（异步 I/O）。
    • 安装方法（参考之前的说明）：bash 

      .\vcpkg install boost-beast:x64-windows
.\vcpkg integrate install
  • 项目配置：
    • Visual Studio 2022 中，添加 Boost 头文件和库路径。
    • C++17 或更高标准，链接 Ws2_32.lib。
  • INetwork 接口：cpp 

    #ifndef NETWORK_INTERFACE_H
#define NETWORK_INTERFACE_H

#include <string>
#include <functional>

class INetwork {
public:
    virtual ~INetwork() = default;
    virtual bool Initialize(const std::string& ip, int port) = 0;
    virtual void Send(const std::string& data, const std::string& targetIp = "", int targetPort = 0,
                     std::function<void(bool, const std::string&)> callback = nullptr) = 0;
    virtual void Receive(std::function<void(const std::string&, const std::string&, int)> callback) = 0;
    virtual void Close() = 0;
    virtual std::string GetError() const = 0;
};

#endif

2. AsyncWebSocketServer 实现AsyncWebSocketServer.h：cpp

#ifndef ASYNC_WEBSOCKET_SERVER_H
#define ASYNC_WEBSOCKET_SERVER_H

#include "NetworkInterface.h"
#include <boost/beast.hpp>
#include <boost/asio.hpp>
#include <memory>
#include <string>
#include <list>

namespace beast = boost::beast;
namespace websocket = beast::websocket;
using tcp = boost::asio::ip::tcp;

class AsyncWebSocketServer : public INetwork {
public:
    AsyncWebSocketServer();
    ~AsyncWebSocketServer() override;

    bool Initialize(const std::string& ip, int port) override;
    void Send(const std::string& data, const std::string& targetIp = "", int targetPort = 0,
              std::function<void(bool, const std::string&)> callback = nullptr) override;
    void Receive(std::function<void(const std::string&, const std::string&, int)> callback) override;
    void Close() override;
    std::string GetError() const override;

private:
    class Connection; // 前向声明连接类
    void StartAccept();

    boost::asio::io_context io_context_;
    std::unique_ptr<tcp::acceptor> acceptor_;
    std::list<std::shared_ptr<Connection>> connections_; // 存储客户端连接
    std::string error_msg_;
    std::function<void(const std::string&, const std::string&, int)> receive_callback_;
};

#endif

AsyncWebSocketServer.cpp：cpp

#include "AsyncWebSocketServer.h"
#include <boost/beast.hpp>
#include <boost/asio.hpp>
#include <iostream>

using tcp = boost::asio::ip::tcp;

// 每个客户端的 WebSocket 连接
class AsyncWebSocketServer::Connection : public std::enable_shared_from_this<Connection> {
public:
    Connection(boost::asio::io_context& io_context)
        : ws_(io_context), strand_(io_context) {}

    websocket::stream<tcp::socket>& GetWebSocket() { return ws_; }

    void Start(std::function<void(const std::string&, const std::string&, int)> callback) {
        // 异步执行 WebSocket 握手
        ws_.async_accept(
            boost::asio::bind_executor(strand_,
                [this, self = shared_from_this(), callback](const boost::system::error_code& error) {
                    if (!error) {
                        StartRead(callback);
                    } else {
                        std::cerr << "WebSocket accept error: " << error.message() << std::endl;
                    }
                }));
    }

    void Send(const std::string& data, std::function<void(bool, const std::string&)> callback) {
        ws_.async_write(
            boost::asio::buffer(data),
            boost::asio::bind_executor(strand_,
                [callback](const boost::system::error_code& error, std::size_t /*bytes_transferred*/) {
                    if (callback) {
                        callback(!error, error ? error.message() : "");
                    }
                }));
    }

    std::string GetClientIp() const {
        return ws_.next_layer().remote_endpoint().address().to_string();
    }

    int GetClientPort() const {
        return ws_.next_layer().remote_endpoint().port();
    }

private:
    void StartRead(std::function<void(const std::string&, const std::string&, int)> callback) {
        ws_.async_read(
            buffer_,
            boost::asio::bind_executor(strand_,
                [this, self = shared_from_this(), callback](const boost::system::error_code& error, std::size_t /*bytes_transferred*/) {
                    if (!error) {
                        std::string data = beast::buffers_to_string(buffer_.data());
                        std::string client_ip = GetClientIp();
                        int client_port = GetClientPort();
                        callback(data, client_ip, client_port);
                        buffer_.consume(buffer_.size());
                        StartRead(callback); // 继续读取
                    } else if (error != beast::websocket::error::closed) {
                        std::cerr << "WebSocket read error: " << error.message() << std::endl;
                    }
                }));
    }

    websocket::stream<tcp::socket> ws_;
    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
    beast::multi_buffer buffer_;
};

AsyncWebSocketServer::AsyncWebSocketServer() : acceptor_(nullptr) {}

AsyncWebSocketServer::~AsyncWebSocketServer() {
    Close();
}

bool AsyncWebSocketServer::Initialize(const std::string& ip, int port) {
    try {
        acceptor_ = std::make_unique<tcp::acceptor>(
            io_context_, tcp::endpoint(boost::asio::ip::address::from_string(ip), port));
        StartAccept();
        return true;
    } catch (const std::exception& e) {
        error_msg_ = e.what();
        return false;
    }
}

void AsyncWebSocketServer::StartAccept() {
    auto connection = std::make_shared<Connection>(io_context_);
    acceptor_->async_accept(
        connection->GetWebSocket().next_layer(),
        [this, connection](const boost::system::error_code& error) {
            if (!error) {
                connections_.push_back(connection);
                connection->Start(receive_callback_);
                StartAccept(); // 继续接受新连接
            } else {
                error_msg_ = error.message();
                std::cerr << "Accept error: " << error_msg_ << std::endl;
            }
        });
}

void AsyncWebSocketServer::Send(const std::string& data, const std::string& targetIp, int targetPort,
                                std::function<void(bool, const std::string&)> callback) {
    bool sent = false;
    for (auto it = connections_.begin(); it != connections_.end();) {
        auto& conn = *it;
        try {
            std::string client_ip = conn->GetClientIp();
            int client_port = conn->GetClientPort();
            if (targetIp.empty() || (client_ip == targetIp && client_port == targetPort)) {
                conn->Send(data, callback);
                sent = true;
            }
            ++it;
        } catch (const std::exception& e) {
            // 客户端断开，移除连接
            it = connections_.erase(it);
        }
    }
    if (!sent && callback) {
        callback(false, "No matching client found");
    }
}

void AsyncWebSocketServer::Receive(std::function<void(const std::string&, const std::string&, int)> callback) {
    receive_callback_ = callback;
    io_context_.run(); // 启动事件循环
}

void AsyncWebSocketServer::Close() {
    if (acceptor_) {
        acceptor_->close();
        acceptor_.reset();
    }
    for (auto& conn : connections_) {
        try {
            conn->GetWebSocket().close(websocket::close_code::normal);
        } catch (const std::exception& e) {
            std::cerr << "Close error: " << e.message() << std::endl;
        }
    }
    connections_.clear();
    io_context_.stop();
}

std::string AsyncWebSocketServer::GetError() const {
    return error_msg_;
}

代码说明：

  • Connection 类：
    • 管理每个客户端的 WebSocket 连接，使用 websocket::stream<tcp::socket>。
    • Start：执行异步握手（async_accept），然后调用 StartRead 读取数据。
    • Send：异步发送 WebSocket 数据帧。
    • strand_：确保线程安全，防止并发访问 socket。
    • GetClientIp 和 GetClientPort：获取客户端 IP 和端口，用于 Send 匹配。
  • AsyncWebSocketServer：
    • Initialize：创建 TCP 接受器，绑定到指定 IP 和端口。
    • StartAccept：异步接受新连接，创建 Connection 对象。
    • Send：向匹配 targetIp 和 targetPort 的客户端发送数据，支持广播（targetIp 为空）。
    • Receive：存储回调函数，启动 io_context_ 事件循环。
    • Close：关闭所有连接和接受器，清理资源。
    • connections_：维护客户端连接列表，支持多客户端管理。

三、main 函数集成以下是修复了 lambda 捕获问题的 main 函数，兼容 WebSocket、TCP 和 UDP：cpp

#include "TcpServer.h"
#include "UdpServer.h"
#include "AsyncWebSocketServer.h"
#include <memory>
#include <iostream>

int main() {
    std::unique_ptr<INetwork> network;
    std::string protocol = "WEBSOCKET"; // 可切换为 "TCP" 或 "UDP"

    // 选择协议
    if (protocol == "TCP") {
        network = std::make_unique<TcpServer>();
    } else if (protocol == "UDP") {
        network = std::make_unique<UdpServer>();
    } else if (protocol == "WEBSOCKET") {
        network = std::make_unique<AsyncWebSocketServer>();
    } else {
        std::cerr << "Unsupported protocol" << std::endl;
        return -1;
    }

    // 初始化服务器
    if (!network->Initialize("127.0.0.1", 8090)) {
        std::cerr << "Initialize failed: " << network->GetError() << std::endl;
        return -1;
    }

    // 异步接收数据
    network->Receive([&network](const std::string& data, const std::string& ip, int port) {
        std::cout << "Received from " << ip << ":" << port << ": " << data << std::endl;
        // 异步发送响应
        network->Send("Server received: " + data, ip, port,
                      [](bool success, const std::string& error) {
                          if (!success) {
                              std::cerr << "Send failed: " << error << std::endl;
                          }
                      });
    });

    return 0;
}

修复说明：

  • lambda 捕获：添加 [&network]，按引用捕获 network，解决编译错误。
  • 异步支持：AsyncWebSocketServer 使用 io_context_.run() 处理事件循环，main 函数无需 while (true)。
  • Send 回调：添加回调函数，处理异步发送的成功或失败。

四、WebSocket 协议细节参考之前的 WebSocket 协议说明，AsyncWebSocketServer 实现涵盖以下关键点：

  1. 握手：
    • ws_.async_accept 处理 HTTP 升级请求，验证 Sec-WebSocket-Key，返回 Sec-WebSocket-Accept。
  2. 数据帧：
    • ws_.async_read 解析 WebSocket 帧（Opcode 0x1 文本帧），解码掩码数据。
    • ws_.async_write 构造帧，无需掩码（服务器端）。
  3. 控制帧：
    • 自动响应 Ping/Pong 帧。
    • Close 方法发送关闭帧（websocket::close_code::normal）。
  4. 多客户端：
    • connections_ 列表存储所有客户端连接，支持定向发送或广播。

五、测试 AsyncWebSocketServer

1. 测试工具

  • wscat（推荐）：bash 

    npm install -g wscat
  • 浏览器（JavaScript）：javascript 

    const ws = new WebSocket("ws://127.0.0.1:8090");
ws.onopen = () => ws.send("Hello, WebSocket!");
ws.onmessage = (e) => console.log("Received: ", e.data);
ws.onerror = (e) => console.error("Error: ", e);
ws.onclose = () => console.log("Connection closed");

2. 测试步骤

  1. 编译和运行：
    • 在 Visual Studio 2022 中编译 main.cpp（设置 protocol = "WEBSOCKET"）。
    • 运行程序，确认监听 127.0.0.1:8090：bash 

      netstat -a | findstr 8090
  2. 连接服务器：
    • 使用 wscat：bash 

      wscat -c ws://127.0.0.1:8090
> Hello, WebSocket!
    • 或在浏览器开发者工具运行上述 JavaScript。
  3. 验证接收：
    • 检查服务器输出： 

      Received from 127.0.0.1:<client_port>: Hello, WebSocket!
  4. 验证发送：
    • wscat 或浏览器应收到： 

      Server received: Hello, WebSocket!
  5. 测试多客户端：
    • 打开多个 wscat 实例，发送不同消息，验证服务器是否正确处理。
    • 检查 connections_ 列表是否包含所有客户端。
  6. 测试关闭：
    • 在 wscat 中按 Ctrl+C，或调用 ws.close()。
    • 验证服务器是否移除断开连接的客户端（connections_.erase）。

3. 使用 SocketTool 测试（有限支持）

  • SocketTool 不直接支持 WebSocket 数据帧，但可测试 HTTP 握手：
    • 配置 TCP 客户端，连接 127.0.0.1:8090。
    • 发送 HTTP 请求：http 

      GET / HTTP/1.1
Host: 127.0.0.1:8090
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
    • 检查服务器是否返回 HTTP 101。
  • 局限性：SocketTool 无法处理 WebSocket 数据帧，建议使用 wscat。

4. 调试

  • 在 Visual Studio 2022 中：
    • 设置断点在 Connection::StartRead 和 Send，检查数据帧和回调。
    • 使用“诊断工具”监控网络流量。
    • 查看 error_msg_ 和 std::cerr 输出，定位错误。

六、注意事项

  1. 异步事件循环：
    • io_context_.run() 驱动所有异步操作，main 函数无需额外循环。
    • 确保 Receive 只调用一次 io_context_.run()，避免重复运行。
  2. 客户端断开处理：
    • Send 方法捕获异常，移除断开连接的客户端。
    • 可添加心跳机制（Ping/Pong）检测客户端活跃状态：cpp 

      void Connection::StartPing() {
    boost::asio::steady_timer timer(ws_.get_executor());
    timer.expires_after(std::chrono::seconds(30));
    timer.async_wait([this, self = shared_from_this()](const boost::system::error_code& error) {
        if (!error) {
            ws_.async_ping({}, [](const boost::system::error_code&) {});
            StartPing();
        }
    });
}
  3. 线程安全：
    • strand_ 确保每个 Connection 的异步操作顺序执行。
    • 多线程支持需多个 io_context：cpp 

      std::vector<std::thread> threads;
for (int i = 0; i < std::thread::hardware_concurrency(); ++i) {
    threads.emplace_back([&io_context] { io_context.run(); });
}
for (auto& t : threads) t.join();
  4. 消息大小：
    • 设置最大消息大小：cpp 

      ws_.read_message_max(64 * 1024 * 1024); // 64MB

七、扩展建议

  1. TLS 支持：
    • 使用 websocket::stream<boost::asio::ssl::stream<tcp::socket>> 支持 wss://：cpp 

      boost::asio::ssl::context ssl_context(boost::asio::ssl::context::tlsv12);
ssl_context.use_certificate_chain_file("server.crt");
ssl_context.use_private_key_file("server.key", boost::asio::ssl::context::pem);
  2. JSON 消息：
    • 集成 nlohmann/json 序列化消息：cpp 

      #include <nlohmann/json.hpp>
nlohmann::json j = {{"message", data}};
conn->Send(j.dump(), callback);
  3. C++20 协程：
    • 使用 co_await 简化异步逻辑：cpp 

      boost::asio::co_spawn(io_context_, [this]() -> boost::asio::awaitable<void> {
    co_await ws_.async_accept(boost::asio::use_awaitable);
    auto [error, bytes] = co_await ws_.async_read(buffer_, boost::asio::use_awaitable);
    if (!error) {
        std::string data = beast::buffers_to_string(buffer_.data());
        receive_callback_(data, GetClientIp(), GetClientPort());
    }
}, boost::asio::detached);
  4. 日志记录：
    • 使用 spdlog：cpp 

      #include <spdlog/spdlog.h>
spdlog::info("Received from {}:{}: {}", client_ip, client_port, data);

八、总结

  • 实现：AsyncWebSocketServer 使用 Boost.Beast 实现异步 WebSocket 通信，兼容 INetwork 接口。
  • lambda 修复：通过 [&network] 捕获解决 network->Send 错误。
  • 测试：使用 wscat 或浏览器验证握手和消息收发，SocketTool 仅限 HTTP 握手测试。
  • 协议细节：处理 HTTP 握手、数据帧和关闭帧，支持多客户端。
  • 参考资料：

如果需要更详细的 TLS 实现、协程支持或测试用例，请告诉我！

# websocket # 服务器 # 网络协议
Logo
智能体开发者社区

中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。

更多推荐

cover

从一张商品图到正式上架：实测店匠 Shoplazza Athena，AI Agent 不只是聊天框

avatar 智能体开发者社区

2026 AI Agent 落地观察：从演示到生产，智能体正在穿越规模化拐点

智能体开始批量走进生产环境。行业调研数据显示，超过半数使用 AI 的企业已将 Agent 部署至生产环境，2026 年被普遍认为是智能体应用的爆发元年——从单点实验走向系统协作，从模型能力竞争转向协议生态竞争。这篇文章梳理近期 Agent 落地的几个关键信号：编程智能体的工作台化、协议层的标准化、企业级 AgentOps 的成型，以及各行业的真实案例，最后聊聊落地里真实的坑。—## 一、格局之变：

avatar 智能体开发者社区
cover

【AI Agent】《2026年9月 AI Agent全栈开发技术选型专项面试宝典》

avatar 智能体开发者社区