RPC框架后端微服务网络通信【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址https://gitcode.com/gh_mirrors/brpc6/brpc点击查看免费下载brpc 的 ExecutionQueue 是一套基于 bthread 的 wait-free MPSC多生产者单消费者异步串行执行队列多个线程可以同时、无锁地提交任务而任务在另一个独立的执行上下文中严格按照提交顺序被消费。它在 brpc 生态中承担着“把对同一资源的并发访问串行化”的职责最典型的生产级用例就是 stream.cpp 中多线程向同一个连接/fd 写数据时的有序消息投递。读完本文你将掌握 ExecutionQueue 的完整 API启动、停止、提交、取消、全部配置参数语义、与 mutex 的选型取舍以及其 wait-free 提交、64 位弱引用 id、任务节点复用等底层实现原理。概述ExecutionQueue 是什么ExecutionQueue 类似于 kylin 的 ExecMan提供“异步串行执行”的能力。这项技术的雏形最早用于在 RPC 中实现多线程向同一个 fd 写数据在 bthread 引入后的某个版本r31345 之后被正式加入 bthread 库。它在 execution_queue.h 中被定义为ExecutionQueue is a special wait-free MPSC queue of which the consumer thread is auto started by the execute operation and auto quits if there are no more tasks。也就是说消费端不是常驻的守护线程而是“有任务时被自动拉起、没有任务时自动退出”的按需模型。它提供如下基本功能异步有序执行任务在另外一个独立的线程bthread中执行且执行顺序严格与提交顺序一致Multi Producer多个线程可以同时向同一个 ExecutionQueue 提交任务支持 cancel可以取消一个已经提交但尚未执行的任务支持 stop优雅停止队列停止后拒绝新任务、排空已提交任务支持高优任务插队高优先级任务按 FIFO 顺序排在所有普通任务之前执行。与 ExecMan 的主要区别提交接口是 wait-free 的ExecMan 依赖 lock当机器整体繁忙时一个进程被系统强制切换可能导致所有等待该锁的线程都被阻塞而 ExecutionQueue 的提交路径无锁见下文“wait-free 提交”一节的源码分析。支持批量处理消费线程一次可以批量处理提交的一批任务获得更好的 cache locality局部性。ExecMan 的某个线程处理完某个 AsyncClient 的 AsyncContext 后下一个任务很可能属于另一个 AsyncClient导致 CPU cache 在不同 AsyncClient 依赖的资源间频繁切换。处理函数不绑定固定线程ExecMan 根据 AsyncClient hash 到固定的执行线程而不同的 ExecutionQueue 之间任务处理完全独立当线程数足够时所有非空闲的 ExecutionQueue 都能同时得到调度。反之线程数不足时 ExecutionQueue 无法保证公平性此时需要动态增加 bthread 的 worker 线程来提升整体处理能力。运行线程是 bthread可以随意使用 bthread 同步原语如 bthread_usleep、butex 等不用担心阻塞 pthread 的执行而在 ExecMan 中要尽量避免使用高概率阻塞的同步原语。设计背景Message Passing 与 Actor 模型在多核并发编程领域Message passing消息传递作为解决竞争的手段应用广泛。它按照业务依赖的资源将逻辑拆分成若干个独立 actor每个 actor 负责对应资源的维护。当一个流程需要修改某个资源时就把这个操作转化为一条消息发送给对应 actoractor通常在另外的执行上下文中根据命令内容对资源进行修改随后选择唤醒调用者同步或提交到下一个 actor异步继续处理。ExecutionQueue 正是这套 actor 模型的落地载体它把“对某个资源的修改”串行化到单一消费者上从而从根本上消除对该资源的并发竞争。当你需要保护的资源只有一个、访问频率又高时把它映射到一个 ExecutionQueue 上、把并发修改转化为有序消息是比直接加锁更可扩展的方案。ExecutionQueue vs Mutex两种并发模型的取舍ExecutionQueue 和 mutex 都可以用来在多线程场景中消除竞争。相比 mutex使用 ExecutionQueue 有这些优点角色划分清晰、概念简单实现中无需考虑锁带来的问题如死锁、锁顺序反转执行顺序有保证任务的执行顺序严格等于提交顺序而 mutex 的唤醒顺序无法得到严格保证线程各司其职所有线程都在做有用的事情不存在为了等待而空转高吞吐在繁忙、卡顿的场景下可以更好地批量执行整体获得较高吞吐。但缺点也同样明显一个流程的代码往往散落在提交方与执行函数多处代码理解和维护成本高为了提高并发度一件事往往被拆分到多个 ExecutionQueue 做流水线处理导致任务在多核之间频繁切换付出额外的调度与 cache 同步开销——尤其是当竞争的临界区非常小时这些开销不可忽略同时原子地操作多个资源变得复杂mutex 可以同时锁住多个锁而 ExecutionQueue 需要依赖额外的 dispatch queue单线程执行模型下某个任务运行慢了会阻塞同一个队列上的其他操作并发控制变复杂队列可能因缓存过多任务而占用过多内存。不考虑性能与复杂度理论上任何系统都可以只用 mutex 或只用 ExecutionQueue 来消除竞争。但复杂系统的设计建议按场景灵活选型临界区非常小、竞争不激烈时优先选择 mutex之后可以结合 contention profiler 判断 mutex 是否成为瓶颈需要严格有序执行或者竞争激烈但可以通过批量执行提升吞吐时选择 ExecutionQueue。特别指出Linux 中 mutex 无竞争的 lock/unlock 只需几条原子指令在绝大多数场景下开销都可以忽略不计。多线程编程没有万能的模型需要结合具体场景与 profiling 工具在复杂度和性能之间找到平衡。快速上手完整使用流程实现执行函数ExecutionQueue 的消费逻辑由用户提供的执行函数承担其签名固定为int (*)(void* meta, TaskIteratorT iter)。TaskIterator支持迭代器式遍历一次回调可能携带一批任务// Iterate over the given tasks // // Example: // // #include bthread/execution_queue.h // // int demo_execute(void* meta, TaskIteratorT iter) { // if (iter.is_queue_stopped()) { // // destroy meta and related resources // return 0; // } // for (; iter; iter) { // // do_something(meta, *iter) // // or do_something(meta, iter-a_member_of_T) // } // return 0; // } template typename T class TaskIterator;几个关键语义见 execution_queue_inl.h 中的operator bool实现iter.is_queue_stopped()返回 true 表示队列已停止且之后永远不会有新任务——这是释放 meta 及相关资源的唯一安全信号之一for (; iter; iter)遍历当前批次内的所有任务*iter或iter-xxx访问任务对象也可以完全不遍历就返回测试用例not_do_iterate_at_all即验证了这种用法但需要注意若返回时迭代器尚未走到末尾clear_before_return 会打印Return a executing node, did you return before iterator reached the end?的警告日志任务节点也会按未迭代完处理。启动一个 ExecutionQueue启动接口execution_queue_start会创建一个队列并把返回的队列句柄写入调用方传入的ExecutionQueueIdT*struct ExecutionQueueOptions { ExecutionQueueOptions(); // Execute in resident pthread instead of bthread. default: false. bool use_pthread; // Attribute of the bthread which execute runs on. default: BTHREAD_ATTR_NORMAL // Bthread will be used when executor NULL and use_pthread false. bthread_attr_t bthread_attr; // Executor that tasks run on. default: NULL // Note that TaskOptions.in_place_if_possible false will not work, if implementation of // Executor is in-place(synchronous). Executor * executor; }; // Start a ExecutionQueue. If |options| is NULL, the queue will be created with // default options. // Returns 0 on success, errno otherwise // NOTE: type |T| can be non-POD but must be copy-constructible template typename T int execution_queue_start( ExecutionQueueIdT* id, const ExecutionQueueOptions* options, int (*execute)(void* meta, TaskIteratorT iter), void* meta);参数细节默认值见 execution_queue_inl.h 的构造函数use_pthread false默认消费任务运行在 bthread 上置为 true 则改用一个常驻 pthread内部通过pthread_create 条件变量_cond唤醒见 execution_queue.cpp。测试用例中对两种模式都做了覆盖test_single_thread、test_performance等均以use_pthread为参数各跑一遍。bthread_attr BTHREAD_ATTR_NORMAL仅当executor NULL use_pthread false时生效用于控制执行 bthread 的属性。executor NULL可注入自定义Executor其唯一接口是virtual int submit(void* (*fn)(void*), void* args) 0见 execution_queue.h队列会通过executor-submit把消费任务交给你的执行器。注意如果 Executor 的实现是同步 in-place 执行则TaskOptions::in_place_if_possible false将不会生效。meta透传给执行函数的用户上下文指针。你必须保证 meta 的生命周期——在对应的 ExecutionQueue 真正停止前不能释放。返回的id是一个 64 位值相当于 ExecutionQueue 实例的一个弱引用可以 wait-free 地在 O(1) 时间内定位队列实例你可以到处拷贝这个 id甚至可以放在 RPC 中作为远端资源的定位工具。64 位 id 的编码方式见下文“源码视角”一节。停止一个 ExecutionQueue// Stop the ExecutionQueue. // After this function is called: // - All the following calls to execution_queue_execute would fail immediately. // - The executor will call |execute| with TaskIterator::is_queue_stopped() being // true exactly once when all the pending tasks have been executed, and after // this point its ok to release the resource referenced by |meta|. // Returns 0 on success, errno othrwise template typename T int execution_queue_stop(ExecutionQueueIdT id); // Wait until the the stop task (Iterator::is_queue_stopped() returns true) has // been executed template typename T int execution_queue_join(ExecutionQueueIdT id);stop与join都可以被多次调用都有合理的行为stop可以随时调用无需担心线程安全性问题。调用stop之后所有后续execution_queue_execute都会立即失败——从实现看ExecutionQueue::execute首先检查stopped()已停止时返回EINVAL见 execution_queue_inl.h外部模板函数在无法定位队列时同样返回EINVALL382-L388。stop会保证当所有已提交任务执行完毕后执行函数会收到恰好一次is_queue_stopped() true的回调此后才可以安全释放 meta 指向的资源。和 fd 的 close 类似如果stop不被调用相应资源会永久泄露——消费线程虽然在任务清空后退出但队列本体及其 slot 不会归还资源池。安全释放 meta 的时机有两种在 execute 函数中收到iter.is_queue_stopped() true的任务时释放或等到join返回后释放。注意不要 double-free。提交任务struct TaskOptions { TaskOptions(); TaskOptions(bool high_priority, bool in_place_if_possible); // Executor would execute high-priority tasks in the FIFO order but before // all pending normal-priority tasks. // NOTE: We dont guarantee any kind of real-time as there might be tasks still // in process which are uninterruptible. // // Default: false bool high_priority; // If |in_place_if_possible| is true, execution_queue_execute would call // execute immediately instead of starting a bthread if possible // // Note: Running callbacks in place might cause the dead lock issue, you // should be very careful turning this flag on. // // Default: false bool in_place_if_possible; }; const static TaskOptions TASK_OPTIONS_NORMAL TaskOptions(/*high_priority*/ false, /*in_place_if_possible*/ false); const static TaskOptions TASK_OPTIONS_URGENT TaskOptions(/*high_priority*/ true, /*in_place_if_possible*/ false); const static TaskOptions TASK_OPTIONS_INPLACE TaskOptions(/*high_priority*/ false, /*in_place_if_possible*/ true); // Thread-safe and Wait-free. // Execute a task with defaut TaskOptions (normal task); template typename T int execution_queue_execute(ExecutionQueueIdT id, typename butil::add_const_referenceT::type task); // Thread-safe and Wait-free. // Execute a task with options. e.g // bthread::execution_queue_execute(queue, task, bthread::TASK_OPTIONS_URGENT) // If |options| is NULL, we will use default options (normal task) // If |handle| is not NULL, we will assign it with the handler of this task. template typename T int execution_queue_execute(ExecutionQueueIdT id, typename butil::add_const_referenceT::type task, const TaskOptions* options); template typename T int execution_queue_execute(ExecutionQueueIdT id, typename butil::add_const_referenceT::type task, const TaskOptions* options, TaskHandle* handle); template typename T int execution_queue_execute(ExecutionQueueIdT id, T task); template typename T int execution_queue_execute(ExecutionQueueIdT id, T task, const TaskOptions* options); template typename T int execution_queue_execute(ExecutionQueueIdT id, T task, const TaskOptions* options, TaskHandle* handle);参数与返回值要点全部提交接口都是Thread-safe 且 Wait-free的。options NULL时使用默认选项普通任务。handle非空时会被赋值为该任务的句柄供后续execution_queue_cancel使用。除按const T提交的重载外还提供了T右值重载支持移动语义。任务类型T可以是非 POD但必须可拷贝构造右值重载还要求支持移动测试 test/bthread_execution_queue_unittest.cpp 中的RValue类型用例专门验证了 move-only 类型禁用了拷贝可以正常提交。返回值语义成功返回 0队列已停止或 id 非法返回EINVAL内存分配失败返回ENOMEM见 execution_queue_inl.h 的execute实现。high_priority任务之间的执行顺序也严格遵循提交顺序这点与 ExecMan 不同ExecMan 的 QueueExecEmergent 的 AsyncContex 执行顺序是 undefined。代价是你无法把任何任务插队到一个 high-priority 任务之前执行。另外注意high_priority只保证“先于所有待执行的普通任务”不保证任何实时性——正在执行中且不可中断的任务无法被打断。in_place_if_possible开启后在无竞争场景下可以省去一次线程调度和 cache 同步的开销——提交线程会直接在本地调用执行函数见下节源码分析。但可能造成死锁或递归层数过深例如任务间不停地 ping-pong开启前请确认你的代码中不存在这些问题。取消一个已提交任务/// [Thread safe and ABA free] Cancel the corresponding task. // Returns: // -1: The task was executed or h is an invalid handle // 0: Success // 1: The task is executing int execution_queue_cancel(const TaskHandle h);取消语义需要特别留意返回-1任务已经被执行过或h是无效句柄如默认构造的TaskHandle其node NULL见 execution_queue_inl.h返回0取消成功任务不会再被执行返回1任务正在执行中无法撤销。返回非 0 仅意味着 ExecutionQueue 已经把这个 task 递交给过 execute真实业务逻辑中可能已经把这个 task 缓存到了其他容器里所以这并不代表逻辑上的任务已经结束你需要在自己的业务层保证这一点。取消是线程安全且ABA-free的TaskHandle中携带任务节点指针node与版本号versionTaskNode::cancel会在锁内校验版本号后才修改状态见 execution_queue_inl.h防止节点被复用后旧句柄误取消新任务。测试cancel、cancel_self、random_cancel、cancel_unexecuted_high_priority_task等用例系统性地覆盖了这些场景。源码视角ExecutionQueue 是如何实现的wait-free 提交单次原子交换提交路径的核心在 ExecutionQueue::execute 与 start_execute提交线程分配一个TaskNode把任务对象 placement-new 进节点内存然后执行一次_head.exchange(node, release)execution_queue.cpp把节点压入以_head为栈顶的链表。_head.exchange本身是单条原子指令无锁、无等待wait-free这正是“机器繁忙时提交线程不会因锁而全部阻塞”的根源如果交换回来的prev_head ! NULL说明队列已有人消费把新节点挂到前驱上即可返回消费线程稍后会把整段新链表“反转”并批量执行_more_tasks中的反转逻辑见 execution_queue_inl.h如果prev_head NULL说明当前没有消费线程提交者“抢到”了执行权此时需要拉起一个后台 bthreadbthread_start_background或 pthread 来执行任务。之所以用后台启动而非前台注释里写得很清楚提交点之后的代码可能是紧急操作比如解锁 pthread_mutex隐式上下文切换可能引发未定义行为如死锁。64 位弱引用 id版本号 槽位ExecutionQueueIdT本质上是一个uint64_t value见 execution_queue_inl.h。其编码规则在 execution_queue.cpp低 32 位ResourceIdExecutionQueueBase槽位索引用于在资源池中 O(1) 定位队列实例高 32 位版本号用于在队列被销毁并复用同一槽位时识别“旧 id”。引用计数_versioned_ref同时打包了“版本号”与“引用计数”两个字段高 32 位版本、低 32 位引用数dereference()在引用数降到 0 时用 CAS 把版本推进到id_ver 2配合_join_butex实现 join 的唤醒见 execution_queue_inl.h。这套机制保证了即使某个线程还持有旧 id它既无法在队列回收后误用内存也不会把已回收的槽位重复归还。性能提示execution_queue_execute(id, ...)每次调用内部都会先execution_queue_address(id)获取一次队列引用、结束时释放这会带来 2 次额外的 cache 更新。在极端性能敏感的场景可以在每个生产者线程开头调用一次execution_queue_address(id)拿到ExecutionQueueT::scoped_ptr_t这是一个智能指针析构时自动dereference之后直接用ptr-execute(...)提交避免每次提交都解析 id见 execution_queue.h 与测试中的push_thread_which_addresses_execq。但注意这会使用户层面的 stop 语义变复杂——只有当没有任何引用时stop 任务才会被传给 execute。若不确定引用所有权不要使用该函数。TaskNode 与任务内存分配TaskNode被定义为BAIDU_CACHELINE_ALIGNMENTcacheline 对齐实测sizeof(TaskNode)为 128 字节见 test/bthread_execution_queue_unittest.cpp其内部结构execution_queue_inl.h包括status任务三态UNEXECUTED / EXECUTING / EXECUTED配合TaskNode::mutex保护是 cancel 逻辑的基础version节点复用版本号供TaskHandle做 ABA 防护high_priority/in_place任务属性快照static_task_mem[56]内嵌 56 字节小对象存储。当sizeof(T) 56时任务对象直接 placement-new 在节点内零额外分配超过 56 字节才走malloc见 TaskAllocatorBase 的大小特化。节点通过butil::ResourcePool分配并复用执行完的任务节点会归还资源池return_task_node这也是“任务缓存过多会占用过多内存”这一缺点的来源——高水位时队列里堆积的节点无法归还给系统。消费线程的启动与退出消费循环主体是 ExecutionQueueBase::_execute_tasksbthread 模式与_execute_tasks_pthreadpthread 模式内部用_mutex_cond等待唤醒。核心流程从头节点开始把链表中的任务分批喂给execute函数一次_execute尽量消费一整批实现批量处理与 cache locality每轮通过_more_tasks检查是否又有新任务被exchange进来有则反转链表继续消费没有则退出消费线程——这就是“无任务时消费线程自动退出”的实现遇到高优任务时优先处理高优子链表见下节队列 stop 后消费线程最后会执行一次带is_queue_stopped() true的调用随后把队列槽位归还资源池。高优任务插队机制提交高优任务时start_execute会先对_high_priority_tasks计数器fetch_add(1)再把节点压栈见 execution_queue.cpp。消费端在每次循环开头检查该计数器大于 0 时以high_priority true调用_execute把所有待处理的高优任务作为一个子链表优先执行执行完nexecuted个就fetch_sub(nexecuted)高优任务之间保持 FIFO 顺序若计数大于 0 但队列里暂时没有高优节点提交者刚fetch_add尚未入栈消费线程会sched_yield()让出 CPU 等待execution_queue.cpp。in_place 立即执行当任务带有in_place属性且当前无人消费时start_execute会直接在提交线程内调用_execute(node, ...)execution_queue.cpp省去一次 bthread 调度。若执行完发现还有更多任务_more_tasks返回 true仍需拉起后台线程继续消费测试should_start_new_thread_on_more_tasks验证了“in-place 执行期间又有新任务时会启动新线程”的行为。bvar 观测指标队列在 execution_queue.cpp 中注册了三个 bvar 指标可在 brpc 内置监控页观察bthread_execq_running_task_count正在运行的任务数bthread_execq_count队列总数bthread_execq_active_count活跃有任务在执行的队列数。测试与真实应用单元测试覆盖test/bthread_execution_queue_unittest.cpp 对 ExecutionQueue 的所有核心行为做了系统性验证几乎每个用例都以use_pthread {false, true}双模式运行覆盖单线程基本流程、move-only 类型提交rvalue、多生产者严格有序multi_threaded_order12 个线程各推 10 万任务验证无乱序、高优插队execute_urgent、urgent_task_is_the_last_task、in-place 执行与再调度、cancel 全语义cancel、cancel_self、random_cancel、cancel_unexecuted_high_priority_task、以及吞吐基准performance会输出每次execution_queue_execute的平均耗时。这些测试既是对 API 契约的说明也是学习每种能力正确用法的现成范例。生产级应用stream 的消息消费串行化ExecutionQueue 在 brpc 内部最典型的应用是 src/brpc/stream.cppstream 连接streaming RPC的消费端用一个ExecutionQueue串行化消息投递——execution_queue_start(s-_consumer_queue, ...)启动消费队列L102接收线程把消息execution_queue_execute(_consumer_queue, tmp)提交进去L471由唯一的消费函数按序处理stream 的定时/超时机制同样借助该队列投递TIMEOUT_TASK哨兵任务L623。这正是“多线程向同一个 fd/连接写数据、串行消费”这一设计初衷的落地。使用注意事项与最佳实践综合文档与源码实践中需要特别留意以下几点必须 stop否则资源泄露像 fd 的 close 一样不调用execution_queue_stop会导致队列资源永久不回收。meta 生命周期与 double-freemeta 可以在执行函数收到is_queue_stopped() true时释放也可以等join返回后释放但绝不能两处都释放。类型要求T必须可拷贝构造move-only 类型请走T重载并确保支持移动。in-place 双刃剑TASK_OPTIONS_INPLACE能省调度开销但有死锁与递归过深风险开启前必须审计提交路径杜绝在 in-place 回调里再次阻塞等待同一队列或互相 ping-pong。高优任务的边界高优只保证相对顺序先于普通任务、自身 FIFO不是实时调度且无法把任务插到某个高优任务之前。cancel 的语义边界返回非 0 只表示任务已被递交给 execute业务层是否真的结束需自行保证句柄要配合 version 使用提交时通过handle参数获取避免 ABA。公平性依赖线程池线程数不足时队列间无法保证公平高负载下需动态扩容 bthread worker 线程。观测与调优通过bthread_execq_*系列 bvar 观察队列水位与活跃度内存敏感场景注意节点缓存占用极端性能场景可改用execution_queue_address拿引用后直接ptr-execute。ExecutionQueue 用“空间换串行、消息换锁”的思路在 brpc 中支撑起了从 stream 消息投递到多线程写同一 fd 等一系列需要有序化并发访问的场景。理解它的 API 契约与底层 wait-free 实现能帮助你在自己的多线程服务中做出更合理的并发模型选型。赞分享RPC框架后端微服务网络通信【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址https://gitcode.com/gh_mirrors/brpc6/brpc点击查看免费下载相关推荐brpc ExecutionQueue 深度解析wait-free 异步串行任务队列的原理与实战brpc ExecutionQueue 深度解析wait free 异步串行任务队列的原理与实战 brpcbetter RPC的 bthread 库提供了后端RPC框架通信网络brpc ExecutionQueue 深入解析bthread 中的无等待Wait-free异步串行任务队列brpc ExecutionQueue 深入解析bthread 中的无等待Wait free异步串行任务队列 导读 ExecutionQueue 是 br后端RPC框架通信网络深入解析 VueUse useAsyncQueue在 AIRI 项目中编排串行异步任务队列深入解析 VueUse useAsyncQueue在 AIRI 项目中编排串行异步任务队列 useAsyncQueue 是 VueUse 提供的一个异步任务编AI 应用人工智能大模型数字人AI Agent语音前端后端桌面应用移动开发即时通讯3D渲染创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考