企业官网建设流程全解析

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 层，可以概括为"一层实现、两层接口"：

1. bthread 层创建多个 worker 分组：每个分组的处理逻辑与原来的单任务组完全一致，只是被拆成了多个相互独立的 worker 池。
2. bthread 接口层增加 tag 字段：在bthread_attr_t中新增了tag字段，用户通过它指定创建的 bthread 在哪个分组上执行。
3. RPC 层增加 server 级 tag 配置：在brpc::ServerOptions中新增了bthread_tag字段，用于指定某个 server 在哪个 worker 分组上执行。

2.1 类型定义与默认值（src/bthread/types.h）

bthread_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_DEFAULT（0）：tag 0 是默认分组；
• bthread_attr_t中的tag字段默认值为BTHREAD_TAG_INVALID（见 types.h），因此普通 bthread 不设置 tag 时仍然运行在原有全局上下文中。

2.2 bthread 层：tagged worker 的启动回调（src/bthread/task_control.cpp）

task control 在创建带 tag 的 worker 时会调用用户注册的启动函数，见 src/bthread/task_control.cpp：

void (*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_tag（src/brpc/server.h）

server 通过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 的定义如下：

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

客户端分别压测 8002（server1，tag 0）与 8003（server2，tag 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.cpp）：

bthread_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 汇总

其中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）上应该避免。

四、Q&A：分组行为深度问答

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 usage）
• bthread_count（bthread 数量）
• 连接信息（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