execution_queue.md

execution_queue

原理

包装ConcurrentBoundedQueue，实现按需激活MPSC消费者的模式

• 每次生产数据时对待处理计数器原子自增，如果自增前为0，边沿触发，启动消费者线程
• 消费者线程持续消费数据，当无数据可消费时，将计数器值和0交换
• 如果交换得到的值和最近一次消费前记录的值无变化，退出消费者线程
• 否则，进入下一轮消费循环

主要用于支持大量低活队列的情况，节省不活跃的监听消费线程，功能和bthread::ExecutionQueue类似，但是

• 消费者通过Executor接口提交，支持使用自定义线程/协程机制
• 保持生产者wait-free提交的同时，不会在竞争时引起消费者阻塞

用法示例

#include "babylon/concurrent/execution_queue.h"

using ::babylon::ConcurrentExecutionQueue;

// 显式定义一个队列
using Queue = ConcurrentExecutionQueue<T>;
Queue queue;

// 设置队列容量到N
// 消费者采用some_executor执行
// 注册lambda消费函数
queue.initialize(N, some_executor, [] (Queue::Iterator iter, Queue::Iterator end) {
  // 消费区间的数据
  while (iter != end) {
    T& item = *iter;
    do_sth_with(item);
    ++iter;
  }
});

// 生产一个数据，并按需启动后台消费
queue.execute("10086");
...

// 等待当前发布数据全部消费完成
// 注意不包含stop语义，可以反复execute & join
queue.join();