ghcr.io/getsentry/taskbroker:678f9f1c9c6ba4e40505f249d52e1f0caaadbc6f-amd64

Taskbroker

Taskbroker 提供 Kafka 消费者、RPC 接口和在途任务存储，构成 Sentry 异步任务执行的核心引擎。本仓库还包含 python 客户端库。

概览

flowchart LR

Sentry -- produce activation --> k[(Kafka)]
k -- consume messages --> Taskbroker
Taskbroker -- store --> sql[(SQLite)]
Worker -- grpc GetTask --> Taskbroker
Worker -- grpc SetTaskStatus --> Taskbroker

Sentry（未来可能包括其他应用）生成任务激活信息。Taskbroker 消费这些消息并将其存储在 SQLite 数据库中，以避免队头阻塞，支持乱序执行和按任务确认。

Worker 通过 gRPC 与 Taskbroker 通信。主要有两个 gRPC 方法：

• GetTask：Worker 可从 Taskbroker 请求任务。
• SetTaskStatus：Worker 完成任务后，使用 SetTaskStatus 报告任务结果。

构建

Taskbroker 采用 Rust 编写，需要最新稳定版 Rust 编译器。编译命令如下：

cargo build --release

生成的二进制文件位于 target/release/ 目录，同时包含调试信息文件。

开发

构建 taskbroker 需要最新稳定版 Rust 和 protoc。仓库根目录包含 Makefile，其中定义了常用开发命令：

• devenv sync：运行设置任务以创建和配置开发环境。
• make unit-test：运行 Rust 测试。
• make integration-test：运行端到端测试。
• make format：使用 cargo fmt 和 cargo clippy 格式化 Rust 代码。

构建与运行

运行 taskbroker 最简单的方式是使用 cargo。还需要运行 kafka。Kafka 通常通过 https://github.com/getsentry/devservices 管理：

# 使用默认配置运行
cargo run

# 使用指定配置文件运行
cargo run -- -c ./config/config-sentry-dev.yaml

配置

Taskbroker 使用 YAML 文件进行配置，所有可用配置选项可参见 https://github.com/getsentry/taskbroker/blob/main/src/config.rs%E3%80%82

所有配置选项也可通过环境变量定义，使用 TASKBROKER_ 前缀。嵌套值使用 __ 分隔符，例如 TASKBROKER_KAFKA_CLUSTERS__DEFAULT__ADDRESS=127.0.0.1:9092。

Kafka 主题与集群

Kafka 通过两个映射进行配置：kafka_clusters 定义每个集群的 broker（及可选的认证信息），kafka_topics 声明每个主题及其所属集群。必须有且仅有一个主题可消费；其余主题（重试、死信）为 produce_only。死信主题和重试主题必须与维护生产者共享同一个集群。

kafka_deadletter_topic: taskworker-dlq

kafka_clusters:
  default:
    address: 127.0.0.1:9092
    # 可选认证信息，应用于该集群的所有主题：
    # security_protocol: sasl_ssl
    # sasl_mechanism: scram-sha-256
    # sasl_username: ...
    # sasl_password: ...

kafka_topics:
  taskworker:
    cluster: default
    consumer_group: taskworker
  taskworker-retry:
    cluster: default
    consumer_group: taskworker
    produce_only: true
  taskworker-dlq:
    cluster: default
    consumer_group: taskworker
    produce_only: true

旧的扁平结构字段 kafka_cluster / kafka_topic / kafka_consumer_group（以及 kafka_deadletter_* 认证字段）已弃用，但仍受支持。有关如何从旧字段迁移到新格式的说明，请参见 docs/kafka-config-migration.md。

测试

测试套件包括 Rust 编写的单元测试和集成测试，以及 Python 编写的端到端测试脚本。

# 运行单元/集成测试
make unit-test

# 运行端到端测试
make integration-test

代码检查

我们使用最新稳定版的 rustfmt 和 clippy 进行代码格式化和检查：

# 检查格式问题
make style

# 修复格式和检查问题
make format

Python 客户端

更多信息请参见 python 客户端 README。

发布客户端

[!NOTE] 客户端与 broker 分开发布。broker 遵循自托管风格的日历版本号（26.4.0），而客户端使用常规语义化版本号（0.1.8）。不过它们共享标签和发布分支，因此 GitHub 上显示的最新发布版本可能不是您版本号的基准。

1. 使用 https://github.com/getsentry/taskbroker/actions/workflows/release-client.yml
2. 选择“Run workflow”并输入版本号。
3. 此操作将使用 craft + getsentry/publish 将包发布到内部和公共 PyPI。