C++ std::semaphore线程信号通知的问题及替代方案咨询

我正在开发一个多线程程序，其中循环运行的流线程会阻塞/休眠/等待控制线程的指示来执行循环，当用户禁用所有流时该线程可以被暂停。

通常我会用std::condition_variable实现这类功能，但由于需求特性——只要有流被请求就允许循环启动，仅当最后一个流被禁用时才停止——我觉得用条件变量的话会比预期复杂（不算特别复杂，但至少要做更多簿记工作）。在cppreference上看到std::semaphore的文档后，我尝试用它来实现：

信号量也常被用于信号/通知语义而非互斥，通过将信号量初始化为0，使尝试调用acquire()的接收方阻塞，直到通知方调用release(n)来“发信号”。在这方面，信号量可视为std::condition_variable的替代方案，通常性能更优。

这是我第一次用C++20的std::semaphore，写了个简单示例，但没按预期工作：

#include <iostream>
#include <thread>
#include <semaphore>
#include <unistd.h>

std::counting_semaphore streams_sem{0};

void stream_thread(){
  while (1){
    streams_sem.acquire();
    std::cout << "Sending images!\n";
    usleep(500000);
    streams_sem.release();
  }
}

int main(int argc, char const* argv[]){
  std::thread t1(stream_thread);
  t1.detach();
  std::cout << "Main thread starting streams\n";
  streams_sem.release();
  usleep(3e6);
  std::cout << "Main thread stopping streams\n";
  streams_sem.acquire();

  return 0;
}

运行后输出如下：

Main thread starting streams
Sending images!
Sending images!
Sending images!
Sending images!
Sending images!
Sending images!
Main thread stopping streams
Sending images!
Sending images!
... and on

我推测主线程从来没在流线程的release()和acquire()之间的间隙被调度，尽管我原本期望它能在几秒内捕获到信号——毕竟主线程已经阻塞在acquire()上了，而且文档也把信号量和条件变量做了对比。

这引出我的核心问题：

1. 要是在流线程的release()后加个短睡眠（10-100微秒），除了浪费时间外，还有哪些隐患？比如会不会还是没法及时停止线程，导致这个方案不可行？
2. 这是不是信号量的不当/无效用法？有没有办法让它更像带内置计数器的信号/条件变量？

我也想了解实现这个功能的其他方法。我本来想避开条件变量，因为觉得它复杂，但如果有更合适的同步方法，也希望能知道。

--- 编辑：补充信息 ---
实际场景中，这些流本质上始终可用，且访问时已经被独立保护，所以我只需要知道有没有流处于活跃状态（理论上可以用忙循环持续检查每个流是否活跃，但希望找到一种轻量的、无请求时能阻塞的实现方式）。

--- 解答 ---

问题根源分析 #

你的示例里，信号量的用法逻辑有问题：流线程每次acquire()后执行任务，然后release()，接着立刻再次acquire()——这个过程中，主线程的acquire()几乎没机会抢到信号量的许可。因为流线程在release()后会马上尝试重新获取，操作系统的线程调度通常会优先让刚释放资源的线程继续执行，所以主线程一直抢不到许可，导致流线程停不下来。

核心问题解答 #

1. 加短睡眠的隐患

   • 加短睡眠确实能给主线程调度的机会，但这是靠“碰运气”的方式，完全依赖调度器的行为，没法保证100%能及时停止。比如系统负载高的时候，哪怕加了睡眠，主线程也可能还是没被调度，流线程会继续跑。
   • 另外，这种方式会引入不必要的延迟，而且随着流线程数量增加，这种不可靠性会被放大。本质上是用“hack”的方式掩盖逻辑问题，不是可靠的解决方案。

2. 信号量的用法是否恰当
   你对信号量的理解有偏差，当前的用法确实不合适。std::counting_semaphore的核心是许可计数，用来控制并发访问的数量，或者实现生产者-消费者模型，但你的场景是“跟踪是否有活跃流，控制线程启停”，用单个信号量的这种来回acquire/release的方式，没法正确表达“最后一个流被禁用才停止”的语义。

更合适的实现方案 #

方案1：用std::condition_variable（其实没你想的复杂）

你的场景用条件变量其实很清晰，只需要一个原子计数器（记录活跃流数量）和一个条件变量：

#include <iostream>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <atomic>
#include <unistd.h>

std::atomic<int> active_streams = 0;
std::condition_variable cv;
std::mutex mtx;
bool stop_thread = false;

void stream_thread() {
    while (!stop_thread) {
        std::unique_lock<std::mutex> lock(mtx);
        // 等待有活跃流，或者线程要停止
        cv.wait(lock, []{ return active_streams > 0 || stop_thread; });
        
        if (stop_thread) break;
        
        // 执行流任务
        std::cout << "Sending images!\n";
        lock.unlock();
        usleep(500000);
    }
}

int main() {
    std::thread t1(stream_thread);
    
    std::cout << "Main thread starting streams\n";
    active_streams++;
    cv.notify_one();
    
    usleep(3e6);
    
    std::cout << "Main thread stopping streams\n";
    active_streams--;
    stop_thread = true;
    cv.notify_one();
    
    t1.join();
    return 0;
}

这个逻辑里，主线程控制active_streams的计数，流线程等待条件变量，当有活跃流时执行任务，没有时阻塞。需要停止线程时设置stop_thread标志即可。

方案2：用std::atomic+自旋等待（轻量但适合低延迟场景）

如果你的场景对延迟要求极高，不想用互斥锁，可以用原子变量配合短时间自旋：

#include <iostream>
#include <thread>
#include <atomic>
#include <unistd.h>

std::atomic<int> active_streams = 0;
std::atomic<bool> stop_thread = false;

void stream_thread() {
    while (!stop_thread) {
        // 自旋等待有活跃流，或者线程停止
        while (active_streams == 0 && !stop_thread) {
            // 让出CPU，减少空转消耗
            std::this_thread::yield();
        }
        
        if (stop_thread) break;
        
        std::cout << "Sending images!\n";
        usleep(500000);
    }
}

int main() {
    std::thread t1(stream_thread);
    
    std::cout << "Main thread starting streams\n";
    active_streams++;
    
    usleep(3e6);
    
    std::cout << "Main thread stopping streams\n";
    active_streams--;
    stop_thread = true;
    
    t1.join();
    return 0;
}

这种方式没有互斥锁的开销，但自旋会消耗CPU，适合活跃流频繁切换的场景，如果流长时间处于禁用状态，还是条件变量更省电。

方案3：修正信号量的用法

如果一定要用信号量，可以结合原子计数器来跟踪活跃流状态，用信号量做通知：

#include <iostream>
#include <thread>
#include <semaphore>
#include <atomic>
#include <unistd.h>

std::binary_semaphore sem{0};
std::atomic<int> active_streams = 0;
std::atomic<bool> stop_thread = false;

void stream_thread() {
    while (!stop_thread) {
        // 等待信号或超时检查停止标志
        if (!sem.try_acquire_for(std::chrono::milliseconds(100))) {
            continue;
        }
        
        if (stop_thread) break;
        
        // 只要有活跃流就循环执行任务
        while (active_streams > 0 && !stop_thread) {
            std::cout << "Sending images!\n";
            usleep(500000);
        }
    }
}

int main() {
    std::thread t1(stream_thread);
    
    std::cout << "Main thread starting streams\n";
    active_streams++;
    sem.release();
    
    usleep(3e6);
    
    std::cout << "Main thread stopping streams\n";
    active_streams--;
    stop_thread = true;
    sem.release(); // 唤醒线程让它检查停止标志
    
    t1.join();
    return 0;
}

这里用二元信号量做通知，原子计数器跟踪活跃流数量，逻辑比你最初的示例更清晰，也能正确实现需求。

内容的提问来源于stack exchange，提问作者Douglas B