C++ 中如何同时等待多个事件并处理最先完成的一个？
如果接口只提供阻塞的 recvMove()，可以为每个邻居创建一个接收线程，把接收结果放进共享队列，由处理线程等待并逐条处理。处理完邻居 x 的输入后，再通知它的接收线程执行下一次 recvMove()。对于最多 9 个客户端的规模，这种做法比较直接，也不需要暴露底层 socket。
如果能调整接口，可以把 poll 和消息解析统一放到 I/O 层，业务代码仍然不接触 socket。std::async 也能启动接收任务，但 C++17 标准库没有 wait_any 接口，无法直接等待一组 std::future 中的任意一个完成。
为什么不能直接等待多个 recvMove()
poll 监听文件描述符的就绪状态，普通成员函数调用没有可供它监听的句柄。
例如：
auto a = neighbors[0]->recvMove();
auto b = neighbors[1]->recvMove();第二次调用要等第一次返回后才会开始。即使给每个 socket 设置一秒超时，接收仍然是串行的。
socket 可读也不代表已经收到完整的 MoveEvent。对于 TCP，收到的数据可能只是消息的一部分。如果 recvMove() 需要多次读取才能组装一条消息，就可能在 socket 首次可读后继续等待。recv(2) 接收语义
保留阻塞接口：每个邻居一个接收线程
接收和处理按以下顺序进行：
- 每个邻居的接收线程执行一次
recvMove()。 - 接收成功后，把邻居编号和消息放进完成队列。
- 处理线程通过
std::condition_variable等待队列非空。 - 取出并处理一条消息，然后允许对应邻居再次接收。
同一个邻居始终只有一次接收在进行。处理某条消息时，其他邻居已经启动的接收会继续执行。
接收接口还需要明确如何表示超时、断开和错误，原来的 MoveEvent recvMove() 没有给出这些约定。可以在项目内实现一个适配函数：
std::optional<MoveEvent> recvMoveOrTimeout(InavjagaGSPIO& peer);返回结果约定如下：
- 返回有值的
std::optional：收到完整消息。 - 返回
std::nullopt：本次等待超时，尚无完整消息。 - 抛出异常：连接断开或发生不可恢复的错误。
适配函数需要根据现有 recvMove() 的行为实现。如果底层已经读到半条消息，超时后应保留解析状态，让下一次调用继续使用。
以下调度示例使用 C++17。MoveEvent 和 InavjagaGSPIO 使用项目中的定义；传入的邻居指针必须有效，且不能重复指向同一条接收连接。
#include <condition_variable>
#include <cstddef>
#include <exception>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <thread>
#include <utility>
#include <vector>
// 项目内适配函数，需要按实际接收接口实现。
std::optional<MoveEvent> recvMoveOrTimeout(InavjagaGSPIO& peer);
struct Completion {
std::size_t neighbor;
std::optional<MoveEvent> move;
std::exception_ptr error;
};
// handle(neighbor, move) 返回 true 表示继续，false 表示结束。
template<class Handler>
void runMoves(
std::vector<std::shared_ptr<InavjagaGSPIO>> neighbors,
Handler handle)
{
if (neighbors.empty()) {
return;
}
std::mutex mutex;
std::condition_variable cv;
std::queue<Completion> completed;
std::vector<bool> armed(neighbors.size(), true);
std::vector<std::thread> workers;
bool stopping = false;
auto finish = [&] {
{
std::lock_guard<std::mutex> lock(mutex);
stopping = true;
}
cv.notify_all();
for (auto& worker : workers) {
if (worker.joinable()) {
worker.join();
}
}
};
try {
workers.reserve(neighbors.size());
for (std::size_t i = 0; i < neighbors.size(); ++i) {
workers.emplace_back([&, i, peer = neighbors[i]] {
for (;;) {
{
std::unique_lock<std::mutex> lock(mutex);
cv.wait(lock, [&] {
return stopping || armed[i];
});
if (stopping) {
return;
}
armed[i] = false;
}
Completion result{i, std::nullopt, {}};
try {
result.move = recvMoveOrTimeout(*peer);
} catch (...) {
result.error = std::current_exception();
}
const bool failed = bool(result.error);
{
std::lock_guard<std::mutex> lock(mutex);
if (stopping) {
return;
}
if (!result.move && !failed) {
// 超时：检查停止状态后，继续等待这个邻居。
armed[i] = true;
} else {
completed.push(std::move(result));
}
}
cv.notify_all();
if (failed) {
return;
}
}
});
}
for (;;) {
Completion result{};
{
std::unique_lock<std::mutex> lock(mutex);
cv.wait(lock, [&] {
return !completed.empty();
});
result = std::move(completed.front());
completed.pop();
}
if (result.error) {
std::rethrow_exception(result.error);
}
// 接收和业务处理期间都不持有队列锁。
if (!handle(result.neighbor, std::move(*result.move))) {
break;
}
{
std::lock_guard<std::mutex> lock(mutex);
armed[result.neighbor] = true;
}
cv.notify_all();
}
} catch (...) {
finish();
throw;
}
finish();
}调用示例：
runMoves(neighbors, [](std::size_t neighbor, MoveEvent move) {
processMove(neighbor, std::move(move));
return true;
});armed 控制每个邻居何时再次接收：处理完 x 的输入，才启动 x 的下一次接收。这样，每个邻居最多保留一条尚未处理完的消息，接收线程不会持续向队列追加该邻居的后续消息。
示例中，任意连接发生不可恢复的错误都会结束整个调度。实际项目也可以改为移除故障邻居，继续服务其他连接。
“最先完成”应如何理解
完成队列按结果入队的顺序处理消息。多个接收几乎同时完成时，线程调度和获取互斥锁的顺序会影响入队次序。
任何邻居完成接收后，都能通知处理线程，无须等待另一个尚未返回的接收。不过，这个顺序不能严格还原网络数据到达的时间顺序。
poll 也可能一次报告多个 socket 就绪，并不提供这些就绪事件的到达顺序。如果业务要求跨连接严格排序，需要在协议中设计序号或其他排序规则。
std::async 可以使用，但仍需完成通知机制
显式指定 std::launch::async，可以让接收调用并发执行：
auto future = std::async(
std::launch::async,
[peer] {
return peer->recvMove();
});默认启动策略允许使用 std::launch::deferred，因此省略策略参数不能保证任务立即并发执行。std::async 语义
std::future::wait()、wait_for() 和 get() 都只针对单个 future。依次调用各个 future 的 get() 时，只要遇到未完成的任务就会阻塞，即使后面的任务已经完成。std::future 接口
连接较少时，可以循环用 wait_for(0ms) 检查各个 future，并适当休眠以避免忙等。这样会引入检查间隔带来的延迟；多个 future 都已就绪时，处理顺序还取决于扫描顺序。使用共享完成队列，则可以在结果产生时直接唤醒处理线程。
未完成任务的 future 也需要保留。释放 std::async 创建的异步共享状态时，可能会等待任务结束，因此丢弃 future 无法取消阻塞中的接收。
可以调整接口时，把多路等待放进 I/O 层
可以在封装内部使用 poll，由 InavjagaIO 负责多路等待，通过友元关系、内部访问接口或同层管理组件获取文件描述符，业务代码仍只接收 MoveEvent。
I/O 层通常需要完成以下工作：
- 为所有连接建立
pollfd数组，一次等待多个连接。 - socket 就绪后，以非阻塞方式读取数据，并更新各连接的解析缓冲区。
- 解析出完整消息后，把消息放入待处理队列。
- 检查
POLLERR、POLLHUP、POLLNVAL，并处理读取时的 EOF 和错误。
poll 可以监听阻塞 socket，但事件循环通常配合非阻塞读取，避免一次读取或消息组装阻塞其他连接。POLLIN 与 POLLHUP 同时出现时，连接中可能还有未读数据。poll(2) 事件说明
poll 调用中的数组名和参数应写为：
struct pollfd pollFds[2] = {};
pollFds[0].fd = socket_0;
pollFds[1].fd = socket_1;
pollFds[0].events = POLLIN;
pollFds[1].events = POLLIN;
int rc = poll(pollFds, 2, 100);pollFds 会转换为指向首元素的指针，这里不应传入 &pollFds。
使用 Libevent 一类事件库时，callback 在事件发生后处理数据，I/O 层仍需负责消息拆分、连接状态和非阻塞接收。把阻塞的 recvMove() 放进 callback，仍可能阻塞整个事件循环。
接收超时和退出需要单独设计
以 Linux 为例，SO_RCVTIMEO 限制底层接收调用的阻塞等待时间。超时且没有传输数据时，调用通常返回 -1，并将 errno 设为 EAGAIN 或 EWOULDBLOCK。它不会生成有效的 MoveEvent，也不会改变 poll 自身的超时设置。socket(7) 超时说明
如果 recvMove() 内部反复调用 recv()，或者遇到超时后自行重试，即使把 SO_RCVTIMEO 设为一秒，也不能保证整个 recvMove() 在一秒内返回。要限制整条消息的接收时间，需要在封装层维护总截止时间。
上述线程示例的停止操作只能唤醒正在等待条件变量的线程，无法中断正在执行的 recvMoveOrTimeout()，因此依赖适配函数在有限时间内返回。如果需要立即停止，I/O 对象应提供能唤醒底层接收的取消接口，再等待线程退出。
共享队列的锁只保护调度状态。其他线程访问同一 I/O 对象时，仍须遵守该对象的线程安全约定。
备注：内容仅供参考。