brpc Bthread Tagged Task Group 完全指南基于 Tag 的线程池隔离、动态调参与监控【免费下载链接】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/GitHub_Trending/brpc/brpc导读在 brpc 中Bthread Tagged Task Group带标签的任务组允许你将 bthread 的 worker 线程池按 tag 划分成多个互不干扰的分组是解决线程资源隔离问题的官方方案。本文以官方文档 docs/en/bthread_tagged_task_group.md 为主体结合 示例代码 与 bthread 源码系统讲解 tag 分组的设计动机、核心 API、启动参数、动态调整线程数的方法以及按 tag 拆分的监控指标帮助你为控制面/数据面隔离、多磁盘隔离、后台任务专用线程池等场景落地一套可复用的实践方案。一、为什么需要线程资源隔离很多应用在开发过程中都会产生线程资源隔离的需求典型场景包括控制面与数据面分离服务分为控制层和数据层数据层请求压力大不希望数据层的流量饿死控制层的处理能力多磁盘隔离服务挂载多块磁盘希望服务不同磁盘的线程之间互不影响避免某一块磁盘的慢请求拖累其他磁盘后台任务与定时任务独立调度一些后台任务或定时任务没有对应的 service却需要在一个独立的线程池中调度其并发度、任务数等都需要由用户自己控制。bthread 的**任务组打标签tagged task group**就是为此设计的将 bthread 的 worker 线程池按照 tag 分组让不同分组之间互不影响达到资源隔离的目的。在此基础上用户还可以叠加更多自定义策略例如将某个 tag 分组的线程限制在指定的 NUMA 节点上运行为某个 tag 分组设置线程本地变量thread-local variables。注意tag 分组是按 server 级别进行的。需要把不同分组的 service 部署到不同的 server 上不同 server 监听不同的端口这一点在示例中体现得非常明显。二、实现原理从 bthread 层到 RPC 层的三层设计从源码结构看tag 分组能力贯穿了 bthread 层与 RPC 层可以概括为一层实现、两层接口bthread 层创建多个 worker 分组每个分组的处理逻辑与原来的单任务组完全一致只是被拆成了多个相互独立的 worker 池。bthread 接口层增加 tag 字段在bthread_attr_t中新增了tag字段用户通过它指定创建的 bthread 在哪个分组上执行。RPC 层增加 server 级 tag 配置在brpc::ServerOptions中新增了bthread_tag字段用于指定某个 server 在哪个 worker 分组上执行。2.1 类型定义与默认值src/bthread/types.hbthread_tag_t本质上是int并且定义了三个关键常量见 src/bthread/types.h// bthread tag default is 0 typedef int bthread_tag_t; static const bthread_tag_t BTHREAD_TAG_INVALID -1; static const bthread_tag_t BTHREAD_TAG_DEFAULT 0;BTHREAD_TAG_INVALID-1表示未启用 tag 分组即传统的不分组的全局线程池模式BTHREAD_TAG_DEFAULT0tag 0 是默认分组bthread_attr_t中的tag字段默认值为BTHREAD_TAG_INVALID见 types.h因此普通 bthread 不设置 tag 时仍然运行在原有全局上下文中。2.2 bthread 层tagged worker 的启动回调src/bthread/task_control.cpptask control 在创建带 tag 的 worker 时会调用用户注册的启动函数见 src/bthread/task_control.cppvoid (*g_tagged_worker_startfn)(bthread_tag_t) nullptr; void run_tagged_worker_startfn(bthread_tag_t tag) { if (g_tagged_worker_startfn) { g_tagged_worker_startfn(tag); } }每个带 tag 的 worker 线程启动时都会回调一次传入该 worker 所属的tag这正是实现绑核等初始化逻辑的挂载点。2.3 RPC 层ServerOptions.bthread_tagsrc/brpc/server.hserver 通过brpc::ServerOptions::bthread_tag指定自身运行在哪个 worker 分组上见 src/brpc/server.h。示例 server.cpp 中brpc::ServerOptions options1; options1.idle_timeout_sec FLAGS_idle_timeout_s; options1.max_concurrency FLAGS_max_concurrency; options1.internal_port FLAGS_internal_port1; options1.bthread_tag FLAGS_tag1; // 绑定到 tag1 分组 options1.num_threads FLAGS_num_threads1; // 该 tag 分组的 worker 数量三、使用方式跑通官方示例仓库中的 example/bthread_tag_echo_c 提供了完整可运行的最小示例服务端将 worker 划分为 3 个 tag 分组其中FLAGS_tag1、FLAGS_tag2给两个不同的 echo server 打标签剩下的一个 tag 给后台任务使用。3.1 服务端启动./echo_server -task_group_ntags 3 -tag1 0 -tag2 1 -bthread_concurrency 20 -bthread_min_concurrency 8 -event_dispatcher_num 1-task_group_ntags 3创建 3 个 tag 分组源码默认值为 1见 task_control.cpp-tag1 0/-tag2 1分别为两个 server 指定 tag 0、tag 1-bthread_concurrency 20所有分组线程总数-bthread_min_concurrency 8所有分组线程数的下限-event_dispatcher_num 1单个分组中事件驱动器event dispatcher的数量源码默认值为 1见 task_control.cpp。在示例 server.cpp 中三个 tag 的定义如下Flag默认值含义tag10server1 的 tagtag21server2 的 tagtag32后台任务的 tagnum_threads16server1tag1 分组的 worker 数量num_threads216server2tag2 分组的 worker 数量3.2 客户端启动./echo_client -dummy_port 8888 -server 0.0.0.0:8002 -use_bthread true ./echo_client -dummy_port 8889 -server 0.0.0.0:8003 -use_bthread true客户端分别压测 8002server1tag 0与 8003server2tag 1两个端口-dummy_port用于在客户端侧启动一个 dummy server仅暴露内置服务-use_bthread true表示使用 bthread 发送请求见 client.cpp。3.3 后台任务与启动回调核心 API 实践示例中后台任务与 worker 启动回调的写法非常典型见 server.cpp// 每个 tag 分组的 worker 启动时都会调用一次tag 参数可用于绑核等初始化 static void my_tagged_worker_start_fn(bthread_tag_t tag) { LOG(INFO) run tagged worker start function tag tag; } // 后台任务查询当前 bthread 所在分组的 tag static void* my_background_task(void*) { while (true) { LOG(INFO) run background task tag bthread_self_tag(); bthread_usleep(1000000UL); } return nullptr; }启动后台任务的写法见 server.cppbthread_t tid; bthread_attr_t attr BTHREAD_ATTR_NORMAL; attr.tag FLAGS_tag3; // 指定后台任务运行在 tag 2 分组 bthread_start_background(tid, attr, my_background_task, nullptr);3.4 核心 API 汇总API声明位置作用bthread_attr_t.tagsrc/bthread/types.h指定创建的 bthread 运行在哪个 tag 分组brpc::ServerOptions.bthread_tagsrc/brpc/server.h指定 server 运行在哪个 worker 分组bthread_set_tagged_worker_startfnsrc/bthread/unstable.h注册每个 tagged worker 启动时的初始化回调用于绑核等bthread_self_tag()bthread 接口查询当前 bthread 所在的 tag 分组其中bthread_set_tagged_worker_startfn的实现位于 src/bthread/bthread.cpp本质上是把回调函数指针赋值给全局的g_tagged_worker_startfn之后每个带 tag 的 worker 启动时都会通过run_tagged_worker_startfn触发。3.5 关于 bthread 的 tag 设置关键路径要避免一般情况下应用创建的 bthread不需要设置bthread_attr_t的 tag 字段创建的 bthread 会在当前 tag 上下文中执行。只有当确实需要把 bthread 调度到其他分组时才设置bthread_attr_t.tag为期望的值——但这样做会有一定的性能损失关键路径hot path上应该避免。四、QA分组行为深度问答Q1如何动态改变某个分组的线程数量分组线程数可以按服务灵活设计启动时线程池依据bthread_concurrency初始化如果设置了bthread_min_concurrency则依据bthread_min_concurrency初始化对 server 来说num_threads就是该 tag 对应的 worker 数量见 server.cpp 中options1.num_threads FLAGS_num_threads1运行期可以通过设置FLAGS_bthread_current_tag和FLAGS_bthread_concurrency_by_tag来改变某个分组的线程数。这两个 flag 在 src/bthread/bthread.cpp 中定义DEFINE_int32(bthread_current_tag, BTHREAD_TAG_INVALID, Set bthread concurrency for this tag); DEFINE_int32(bthread_concurrency_by_tag, 8 BTHREAD_EPOLL_THREAD_NUM, Set bthread concurrency for bthread_current_tag);即bthread_current_tag指定要修改哪个 tag 分组的线程数bthread_concurrency_by_tag指定该分组的新线程数。如果没有设置这两个 flag相当于未启用分组bthread_current_tag保持默认的BTHREAD_TAG_INVALID则num_threads的含义是所有分组的 worker 总数。Q2不同分组之间有什么关系不同分组是完全独立的线程池和事件驱动器彼此没有任何交互这正是资源隔离的实现基础。Q3可以在分组之间做 bthread 的同步操作吗可以。每个 bthread 都有自己的 tag 标签bthread 挂起如等待 butex、睡眠等后重新投入运行时会继续在原 tag 对应的线程池上执行因此跨分组的同步操作是安全的不会因为调度到别的分组而破坏语义。Q4客户端发送和接收 RPC 消息是在哪个分组上执行的取决于客户端的上下文如果客户端不在任何 tag 分组上将使用tag 0分组收发消息否则在当前所在的 tag 分组收发消息。Q5如何将一个分组的线程绑定到指定 CPU使用int bthread_set_tagged_worker_startfn(void (*start_fn)(bthread_tag_t))。该函数用于在某个分组上做初始化工作比如实现绑核根据传入的tag参数为不同分组绑定不同的 CPU 集合。实现方式见 src/bthread/bthread.cpp声明见 src/bthread/unstable.h。五、监控按 tag 拆分的指标目前监控按 tag 划分的指标包括线程的数量worker count线程的使用量worker usagebthread_countbthread 数量连接信息connection info下面是官方文档提供的三张监控截图分别展示线程使用量与动态调整线程数后的效果线程使用量按 tag 拆分动态调整 tag1 分组的线程数仅设置 tag1动态调整所有分组的线程数设置所有 tag从图中可以看到监控面板以bthread_worker_usage、bthread_worker_usage_0、bthread_worker_usage_1以及对应的bthread_worker_count系列区分总量与各 tag 分组配合FLAGS_bthread_current_tag/FLAGS_bthread_concurrency_by_tag动态调参可以直观地观察每个分组线程数的变化是否达到预期。六、典型应用场景小结控制面/数据面隔离为控制面服务与数据面服务分配不同的 tag分别部署在不同端口的 server 上即使数据面流量暴涨也不会拖垮控制面多磁盘/多租户隔离为服务不同磁盘的分组打不同 tag磁盘间故障与慢请求互不影响后台任务独立线程池为无 service 的后台/定时任务分配专用 tag 分组并发度完全由自己控制NUMA 亲和与绑核借助bthread_set_tagged_worker_startfn回调按 tag 将不同分组绑定到不同 CPU/NUMA 节点。需要特别注意的是tag 分组是按 server 级别划分的需要把不同分组的 service 放到不同 server不同端口上。如果需要进一步了解 bthread 本身的工作机制可参考 bthread 文档 与 threading overview关于 server 配置的更多选项可参考 server 文档。【免费下载链接】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/GitHub_Trending/brpc/brpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 brpc Bthread Tagged Task Group 完全指南：基于 Tag 的线程池隔离、动态调参与监控
brpc Bthread Tagged Task Group 完全指南基于 Tag 的线程池隔离、动态调参与监控【免费下载链接】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/GitHub_Trending/brpc/brpc导读在 brpc 中Bthread Tagged Task Group带标签的任务组允许你将 bthread 的 worker 线程池按 tag 划分成多个互不干扰的分组是解决线程资源隔离问题的官方方案。本文以官方文档 docs/en/bthread_tagged_task_group.md 为主体结合 示例代码 与 bthread 源码系统讲解 tag 分组的设计动机、核心 API、启动参数、动态调整线程数的方法以及按 tag 拆分的监控指标帮助你为控制面/数据面隔离、多磁盘隔离、后台任务专用线程池等场景落地一套可复用的实践方案。一、为什么需要线程资源隔离很多应用在开发过程中都会产生线程资源隔离的需求典型场景包括控制面与数据面分离服务分为控制层和数据层数据层请求压力大不希望数据层的流量饿死控制层的处理能力多磁盘隔离服务挂载多块磁盘希望服务不同磁盘的线程之间互不影响避免某一块磁盘的慢请求拖累其他磁盘后台任务与定时任务独立调度一些后台任务或定时任务没有对应的 service却需要在一个独立的线程池中调度其并发度、任务数等都需要由用户自己控制。bthread 的**任务组打标签tagged task group**就是为此设计的将 bthread 的 worker 线程池按照 tag 分组让不同分组之间互不影响达到资源隔离的目的。在此基础上用户还可以叠加更多自定义策略例如将某个 tag 分组的线程限制在指定的 NUMA 节点上运行为某个 tag 分组设置线程本地变量thread-local variables。注意tag 分组是按 server 级别进行的。需要把不同分组的 service 部署到不同的 server 上不同 server 监听不同的端口这一点在示例中体现得非常明显。二、实现原理从 bthread 层到 RPC 层的三层设计从源码结构看tag 分组能力贯穿了 bthread 层与 RPC 层可以概括为一层实现、两层接口bthread 层创建多个 worker 分组每个分组的处理逻辑与原来的单任务组完全一致只是被拆成了多个相互独立的 worker 池。bthread 接口层增加 tag 字段在bthread_attr_t中新增了tag字段用户通过它指定创建的 bthread 在哪个分组上执行。RPC 层增加 server 级 tag 配置在brpc::ServerOptions中新增了bthread_tag字段用于指定某个 server 在哪个 worker 分组上执行。2.1 类型定义与默认值src/bthread/types.hbthread_tag_t本质上是int并且定义了三个关键常量见 src/bthread/types.h// bthread tag default is 0 typedef int bthread_tag_t; static const bthread_tag_t BTHREAD_TAG_INVALID -1; static const bthread_tag_t BTHREAD_TAG_DEFAULT 0;BTHREAD_TAG_INVALID-1表示未启用 tag 分组即传统的不分组的全局线程池模式BTHREAD_TAG_DEFAULT0tag 0 是默认分组bthread_attr_t中的tag字段默认值为BTHREAD_TAG_INVALID见 types.h因此普通 bthread 不设置 tag 时仍然运行在原有全局上下文中。2.2 bthread 层tagged worker 的启动回调src/bthread/task_control.cpptask control 在创建带 tag 的 worker 时会调用用户注册的启动函数见 src/bthread/task_control.cppvoid (*g_tagged_worker_startfn)(bthread_tag_t) nullptr; void run_tagged_worker_startfn(bthread_tag_t tag) { if (g_tagged_worker_startfn) { g_tagged_worker_startfn(tag); } }每个带 tag 的 worker 线程启动时都会回调一次传入该 worker 所属的tag这正是实现绑核等初始化逻辑的挂载点。2.3 RPC 层ServerOptions.bthread_tagsrc/brpc/server.hserver 通过brpc::ServerOptions::bthread_tag指定自身运行在哪个 worker 分组上见 src/brpc/server.h。示例 server.cpp 中brpc::ServerOptions options1; options1.idle_timeout_sec FLAGS_idle_timeout_s; options1.max_concurrency FLAGS_max_concurrency; options1.internal_port FLAGS_internal_port1; options1.bthread_tag FLAGS_tag1; // 绑定到 tag1 分组 options1.num_threads FLAGS_num_threads1; // 该 tag 分组的 worker 数量三、使用方式跑通官方示例仓库中的 example/bthread_tag_echo_c 提供了完整可运行的最小示例服务端将 worker 划分为 3 个 tag 分组其中FLAGS_tag1、FLAGS_tag2给两个不同的 echo server 打标签剩下的一个 tag 给后台任务使用。3.1 服务端启动./echo_server -task_group_ntags 3 -tag1 0 -tag2 1 -bthread_concurrency 20 -bthread_min_concurrency 8 -event_dispatcher_num 1-task_group_ntags 3创建 3 个 tag 分组源码默认值为 1见 task_control.cpp-tag1 0/-tag2 1分别为两个 server 指定 tag 0、tag 1-bthread_concurrency 20所有分组线程总数-bthread_min_concurrency 8所有分组线程数的下限-event_dispatcher_num 1单个分组中事件驱动器event dispatcher的数量源码默认值为 1见 task_control.cpp。在示例 server.cpp 中三个 tag 的定义如下Flag默认值含义tag10server1 的 tagtag21server2 的 tagtag32后台任务的 tagnum_threads16server1tag1 分组的 worker 数量num_threads216server2tag2 分组的 worker 数量3.2 客户端启动./echo_client -dummy_port 8888 -server 0.0.0.0:8002 -use_bthread true ./echo_client -dummy_port 8889 -server 0.0.0.0:8003 -use_bthread true客户端分别压测 8002server1tag 0与 8003server2tag 1两个端口-dummy_port用于在客户端侧启动一个 dummy server仅暴露内置服务-use_bthread true表示使用 bthread 发送请求见 client.cpp。3.3 后台任务与启动回调核心 API 实践示例中后台任务与 worker 启动回调的写法非常典型见 server.cpp// 每个 tag 分组的 worker 启动时都会调用一次tag 参数可用于绑核等初始化 static void my_tagged_worker_start_fn(bthread_tag_t tag) { LOG(INFO) run tagged worker start function tag tag; } // 后台任务查询当前 bthread 所在分组的 tag static void* my_background_task(void*) { while (true) { LOG(INFO) run background task tag bthread_self_tag(); bthread_usleep(1000000UL); } return nullptr; }启动后台任务的写法见 server.cppbthread_t tid; bthread_attr_t attr BTHREAD_ATTR_NORMAL; attr.tag FLAGS_tag3; // 指定后台任务运行在 tag 2 分组 bthread_start_background(tid, attr, my_background_task, nullptr);3.4 核心 API 汇总API声明位置作用bthread_attr_t.tagsrc/bthread/types.h指定创建的 bthread 运行在哪个 tag 分组brpc::ServerOptions.bthread_tagsrc/brpc/server.h指定 server 运行在哪个 worker 分组bthread_set_tagged_worker_startfnsrc/bthread/unstable.h注册每个 tagged worker 启动时的初始化回调用于绑核等bthread_self_tag()bthread 接口查询当前 bthread 所在的 tag 分组其中bthread_set_tagged_worker_startfn的实现位于 src/bthread/bthread.cpp本质上是把回调函数指针赋值给全局的g_tagged_worker_startfn之后每个带 tag 的 worker 启动时都会通过run_tagged_worker_startfn触发。3.5 关于 bthread 的 tag 设置关键路径要避免一般情况下应用创建的 bthread不需要设置bthread_attr_t的 tag 字段创建的 bthread 会在当前 tag 上下文中执行。只有当确实需要把 bthread 调度到其他分组时才设置bthread_attr_t.tag为期望的值——但这样做会有一定的性能损失关键路径hot path上应该避免。四、QA分组行为深度问答Q1如何动态改变某个分组的线程数量分组线程数可以按服务灵活设计启动时线程池依据bthread_concurrency初始化如果设置了bthread_min_concurrency则依据bthread_min_concurrency初始化对 server 来说num_threads就是该 tag 对应的 worker 数量见 server.cpp 中options1.num_threads FLAGS_num_threads1运行期可以通过设置FLAGS_bthread_current_tag和FLAGS_bthread_concurrency_by_tag来改变某个分组的线程数。这两个 flag 在 src/bthread/bthread.cpp 中定义DEFINE_int32(bthread_current_tag, BTHREAD_TAG_INVALID, Set bthread concurrency for this tag); DEFINE_int32(bthread_concurrency_by_tag, 8 BTHREAD_EPOLL_THREAD_NUM, Set bthread concurrency for bthread_current_tag);即bthread_current_tag指定要修改哪个 tag 分组的线程数bthread_concurrency_by_tag指定该分组的新线程数。如果没有设置这两个 flag相当于未启用分组bthread_current_tag保持默认的BTHREAD_TAG_INVALID则num_threads的含义是所有分组的 worker 总数。Q2不同分组之间有什么关系不同分组是完全独立的线程池和事件驱动器彼此没有任何交互这正是资源隔离的实现基础。Q3可以在分组之间做 bthread 的同步操作吗可以。每个 bthread 都有自己的 tag 标签bthread 挂起如等待 butex、睡眠等后重新投入运行时会继续在原 tag 对应的线程池上执行因此跨分组的同步操作是安全的不会因为调度到别的分组而破坏语义。Q4客户端发送和接收 RPC 消息是在哪个分组上执行的取决于客户端的上下文如果客户端不在任何 tag 分组上将使用tag 0分组收发消息否则在当前所在的 tag 分组收发消息。Q5如何将一个分组的线程绑定到指定 CPU使用int bthread_set_tagged_worker_startfn(void (*start_fn)(bthread_tag_t))。该函数用于在某个分组上做初始化工作比如实现绑核根据传入的tag参数为不同分组绑定不同的 CPU 集合。实现方式见 src/bthread/bthread.cpp声明见 src/bthread/unstable.h。五、监控按 tag 拆分的指标目前监控按 tag 划分的指标包括线程的数量worker count线程的使用量worker usagebthread_countbthread 数量连接信息connection info下面是官方文档提供的三张监控截图分别展示线程使用量与动态调整线程数后的效果线程使用量按 tag 拆分动态调整 tag1 分组的线程数仅设置 tag1动态调整所有分组的线程数设置所有 tag从图中可以看到监控面板以bthread_worker_usage、bthread_worker_usage_0、bthread_worker_usage_1以及对应的bthread_worker_count系列区分总量与各 tag 分组配合FLAGS_bthread_current_tag/FLAGS_bthread_concurrency_by_tag动态调参可以直观地观察每个分组线程数的变化是否达到预期。六、典型应用场景小结控制面/数据面隔离为控制面服务与数据面服务分配不同的 tag分别部署在不同端口的 server 上即使数据面流量暴涨也不会拖垮控制面多磁盘/多租户隔离为服务不同磁盘的分组打不同 tag磁盘间故障与慢请求互不影响后台任务独立线程池为无 service 的后台/定时任务分配专用 tag 分组并发度完全由自己控制NUMA 亲和与绑核借助bthread_set_tagged_worker_startfn回调按 tag 将不同分组绑定到不同 CPU/NUMA 节点。需要特别注意的是tag 分组是按 server 级别划分的需要把不同分组的 service 放到不同 server不同端口上。如果需要进一步了解 bthread 本身的工作机制可参考 bthread 文档 与 threading overview关于 server 配置的更多选项可参考 server 文档。【免费下载链接】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/GitHub_Trending/brpc/brpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考