C/C++ 中实现类似 Go Channels 功能的方法

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

template <typename T>
class MessageQueue {
public:
    void enqueue(T item) {
        std::unique_lock<std::mutex> lock(mutex_);
        queue_.push(item);
        condition_.notify_one();
    }

    T dequeue() {
        std::unique_lock<std::mutex> lock(mutex_);
        condition_.wait(lock, [this]{ return !queue_.empty(); });
        T item = queue_.front();
        queue_.pop();
        return item;
    }

private:
    std::queue<T> queue_;
    std::mutex mutex_;
    std::condition_variable condition_;
};

// 示例用法
int main() {
    MessageQueue<int> queue;

    // 生产者线程
    std::thread producer([&queue]() {
        for (int i = 0; i < 10; ++i) {
            queue.enqueue(i);
            std::cout << "Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    });

    // 消费者线程
    std::thread consumer([&queue]() {
        for (int i = 0; i < 10; ++i) {
            int item = queue.dequeue();
            std::cout << "Dequeued: " << item << std::endl;
        }
    });

    producer.join();
    consumer.join();

    return 0;
}

#include "ace/Task.h"
#include "ace/Thread_Manager.h"
#include "ace/Message_Queue.h"

class WorkerTask : public ACE_Task<ACE_MT_SYNCH> {
public:
    WorkerTask() : ACE_Task<ACE_MT_SYNCH>(ACE_Thread_Manager::instance()) {}

    int svc() {
        while (true) {
            ACE_Message_Block* mb = nullptr;
            if (msg_queue()->dequeue_msg(mb) == -1) {
                break; // 线程退出
            }

            // 处理消息
            std::string message(mb->rd_ptr(), mb->length());
            std::cout << "Thread " << ACE_Thread::self() << " received: " << message << std::endl;

            mb->release(); // 释放消息块
        }
        return 0;
    }
};

int main() {
    WorkerTask worker;
    worker.activate(THR_NEW_LWP, 4); // 创建 4 个工作线程

    ACE_Message_Queue<ACE_MT_SYNCH>* queue = worker.msg_queue();

    // 发送消息到队列
    for (int i = 0; i < 10; ++i) {
        std::string message = "Message " + std::to_string(i);
        ACE_Message_Block* mb = new ACE_Message_Block(message.length() + 1);
        std::strcpy(mb->wr_ptr(), message.c_str());
        mb->wr_ptr(message.length() + 1);
        queue->enqueue_tail(mb);
    }

    ACE_Thread_Manager::instance()->wait(); // 等待所有线程结束
    return 0;
}