c++ - 如何使用 C++11 <thread> 设计一个从源中提取数据的系统

Question

本题来自: C++11 thread doesn't work with virtual member function

正如评论中所建议的，我在上一篇文章中提出的问题可能不是正确的问题，所以这里是原始问题:

我想制作一个捕获系统，它将以恒定/动态频率(因来源而异，例如 10 次/秒)查询一些源，并将数据拉入每个队列。虽然源不固定，但它们可能会在运行时添加/删除。

还有一个监视器以恒定的频率从队列中提取数据并显示数据。

那么这个问题的最佳设计模式或结构是什么。

我正在尝试为所有源拉取器创建一个列表，每个拉取器都拥有一个线程和一个指定的拉取函数(拉取函数可能以某种方式与拉取器交互，比如如果源是耗尽的，它会询问停止对该线程的拉取过程。)

Answer 1

除非您查询源的操作是阻塞的(或者您有很多)，否则您不需要为此使用线程。我们可以从一个 Producer 开始，它将使用同步或异步(线程)调度:

template <typename OutputType>
class Producer
{
    std::list<OutputType> output;

protected:
    int poll_interval; // seconds? milliseconds?
    virtual OutputType query() = 0;

public:
    virtual ~Producer();

    int next_poll_interval() const { return poll_interval; }
    void poll() { output.push_back(this->query()); }

    std::size_t size() { return output.size(); }
    // whatever accessors you need for the queue here:
    // pop_front, swap entire list, etc.
};

现在我们可以从这个Producer 派生并在每个子类型中实现query 方法。您可以在构造函数中设置 poll_interval 并将其保留，或者在每次调用 query 时更改它。这是您的通用生产者组件，不依赖于调度机制。

template <typename OutputType>
class ThreadDispatcher
{
    Producer<OutputType> *producer;
    bool shutdown;
    std::thread thread;

    static void loop(ThreadDispatcher *self)
    {
        Producer<OutputType> *producer = self->producer;

        while (!self->shutdown)
        {
            producer->poll();
            // some mechanism to pass the produced values back to the owner
            auto delay = // assume millis for sake of argument
                std::chrono::milliseconds(producer->next_poll_interval());
            std::this_thread::sleep_for(delay);
        }
    }

public:
    explicit ThreadDispatcher(Producer<OutputType> *p)
      : producer(p), shutdown(false), thread(loop, this)
    {
    }

    ~ThreadDispatcher()
    {
        shutdown = true;
        thread.join();
    }

    // again, the accessors you need for reading produced values go here
    // Producer::output isn't synchronised, so you can't expose it directly
    // to the calling thread
};

这是一个简单的调度程序草图，它可以在一个线程中运行您的生产者，并根据您的要求经常轮询它。请注意，未显示将生成的值传回所有者，因为我不知道您想如何访问它们。

另请注意，我还没有同步访问关闭标志 - 它可能应该是原子的，但无论您选择如何处理生成的值，它都可能隐式同步。

有了这个组织，编写一个同步调度程序来在单个线程中查询多个生产者也很容易，例如从选择/轮询循环，或者使用类似 Boost.Asio 的东西和每个生产者的截止时间计时器。