shiblon/entroq

EntroQ

一个具有强竞争消费者语义和事务性更新的任务队列。

有关其如何融入系统的文章，请参见：
[***]

此包的Go组件文档可在 entrogo.com/entroq Go包在线文档中找到。

发音为"Entro-Q"（"Entro-Queue"），意为"Entro-P"之后的字母。旨在远离并行系统的混乱，是早期不够健壮的github.com/shiblon/taskstore的继任者。

设计目标是尽可能简单和模块化，专注于做好一件事：仅作为竞争消费者工作队列，不会扩展为pubsub系统、数据库或通用RPC机制，因此能将这一功能做到极致。

使用

Docker容器可在Docker Hub获取：shiblon/entroq。可用于启动EntroQ服务，然后通过提供的Go或Python库与其通信。默认暴露端口37706。

默认服务使用内存工作队列，可选配预写日志（write-ahead log）以实现持久化和容错。另有PostgreSQL后端版本可供选择。

若仅需Go进程内工作队列，可直接使用库的内存实现，无需服务。其他语言需通过gRPC-based语言特定库与服务通信。

提供命令行客户端与EntroQ服务交互：

Go客户端：

bash
go install entrogo.com/entroq/cmd/eqc@latest
eqc --help

Python客户端：

bash
python3 -m pip install git+https://github.com/shiblon/entroq
python3 -m entroq --help

核心概念

EntroQ支持两种原子性修改操作：

• 从一个或多个队列中声明可用任务
• 更新一组任务（删除、修改、插入），可选依赖于其他任务版本的被动存在

另有只读访问操作，如列出队列中的任务、队列大小等。这些操作无事务属性，是队列状态的尽力而为快照，且不会导致核心操作饥饿。

Claim和Modify操作会递增所有受影响任务的版本号。任何任务被修改时版本都会增加，因此若一个进程修改了任务，其他进程对该任务的操作将因版本不匹配而失败。这是"一次提交"语义的核心。

与许多用作竞争消费者队列的pub/sub系统不同，EntroQ消除了交付后工作丢失或重复提交的可能性，核心原则是：
工作提交永不丢失或重复。

任务（Tasks）

任务的基本属性：

• 队列名称（Queue Name）
• 全局唯一ID（Globally Unique ID）
• 版本（Version）
• 到达时间（Arrival Time）
• 值（Value）

任务还包含当前声明客户端ID、创建时间、修改时间、声明次数等元数据（实现细节）。

队列（Queues）

EntroQ支持多队列，但队列并非一级概念，仅在包含任务时存在。若某队列无任务，则该队列不存在。在数据库实现中，队列是任务的属性（如PostgreSQL后端中tasks表的一列）。

任务ID持久不变，仅版本在修改时递增，因此可跟踪单个任务的多次修改。任何修改操作需ID和版本均匹配，否则失败并返回错误任务ID。这确保了竞争消费者工作器不会意外覆盖彼此的任务。

声明（Claims）

Claim操作针对一个或多个队列，随机返回到达时间（at）在过去的可用任务（若有），否则阻塞。另有非阻塞的TryClaim操作（无任务时返回nil）。

指定多个队列时，仅返回其中一个队列的一个任务，支持工作器从多队列公平拉取任务。例如，"快车道"队列可让工作器优先处理高优先级任务，任务从多队列公平拉取，短队列任务会先于长队列被消费。

成功声明任务时，原子性执行：

• 版本递增
• 到达时间设为未来时间点

版本递增确保先前声明者无法更新任务；未来到达时间给新声明者留出处理时间或"到达时间更新"周期（定期续期任务租约），也允许网络传输时间。

最佳实践：初始声明时间宜短，依赖声明者定期更新到达时间。这能快速从网络分区等异常中恢复（如30秒续期窗口错过后，新工作器能更快接手任务，而非等待15分钟租约到期）。

修改（Modify）

Modify调用可原子事务中执行多种修改：

• 插入（Insert）：添加一个或多个任务到指定队列
• 删除（Delete）：删除一个或多个任务
• 变更（Change）：修改任务的值、队列或到达时间
• 依赖（Depend）：要求一个或多个任务及版本被动存在

若任何依赖不存在（如待删除/修改/依赖的任务不存在，或插入时ID冲突），整个操作失败并返回具体错误。

因工作需显式确认（删除）才会丢失，收到依赖错误时可安全放弃当前任务，获取新任务继续处理。EntroQ设计可避免因任务数据导致工作器崩溃而造成的队列饥饿，问题任务会定期重试，其他任务可正常处理。

工作器（Workers）

创建EntroQ客户端实例后，可创建"工作器"——在一个或多个队列上执行声明-续期-修改循环。可在goroutine（或线程）中运行，标准库工作器让用户专注于任务处理逻辑，后台自动续期任务租约。Go提供完善的工作器实现，contrib/py提供Python实现，原理可在任何支持gRPC的语言中实现。

工作器代码负责处理声明的任务，成功后返回预期修改。标准工作器代码会处理任务续期导致的版本增量，确保修改时版本正确。

典型使用流程：启动EntroQ服务，部署一个或多个工作器处理任务流。可通过库或命令行客户端注入初始任务。

最佳实践：避免设计流水线，建议使用"蹦床"（trampoline）工作器：单一全局队列接收响应，根据内容和状态推送到不同任务队列。流水线随规模增长复杂度呈指数级增加，而蹦床工作器使每个工作器"单一用途"，状态转换集中管理。

一次提交，可能多次工作（Commit Once, Maybe Work Twice）

EntroQ任务仅可被"确认"（删除或移动）一次。多个声明者（工作器）可能同时处理同一任务，但只有一个能成功删除任务，其他会因版本不匹配失败并丢弃工作。

因此，实际工作可能执行多次，但仅会被持久化记录一次。需设计幂等工作器，原则如下：

• 所有外部修改需幂等
• 输出文件需每次生成唯一名称

幂等性原则：使用稳定绝对值而非相对更新（如设置最终值而非递增）。
唯一文件名原则：每个工作器（即使处理同一任务）生成随机/时间戳文件名，避免部分写入损坏完整文件，选择垃圾回收而非写入效率。

安全工作准则

通过ID/版本区分可实现"强制删除"等功能（如命令行客户端），但正常工作器代码应仅处理已声明的任务，避免覆盖声明者或版本。
规则：仅处理已声明任务，永不覆盖声明者或版本。
手动干预时使用命令行客户端，确保人工判断潜在影响。

后端实现

Go程序通过entroq.New函数和BackendOpener创建EntroQ实例。Python客户端始终通过gRPC与EntroQ服务通信。

Go有三种主要后端：

• 内存（In-memory）
• PostgreSQL
• gRPC客户端

内存后端（In-Memory Backend）

eqmem后端支持进程内工作队列，API与网络实现一致，便于从内存过渡到数据库/网络实现。

示例代码：

go
package main

import (
  "context"

  "entrogo.com/entroq"
  "entrogo.com/entroq/backend/eqmem"
)

func main() {
  ctx := context.Background()
  eq := entroq.New(ctx, eqmem.Opener())

  // 使用eq.Modify, eq.Insert, eq.Claim等方法（通常在goroutine中）
}

内存后端包含预写日志实现用于持久化，可配置日志目录和轮转策略。Docker镜像默认使用带日志和定期快照的内存实现，启动gRPC服务。

gRPC后端

grpc后端将entroq.EntroQ客户端转换为gRPC客户端，与qsvc实现通信，支持认证等gRPC特性。

示例代码：

go
package main

import (
  "context"

  "entrogo.com/entroq"
  "entrogo.com/entroq/backend/eqgrpc"
)

func main() {
  ctx := context.Background()
  eq := entroq.New(ctx, eqgrpc.Opener(":37706"))

  // 使用eq.Modify, eq.Insert, eq.Claim等方法（通常在goroutine中）
}

Opener支持主机名及mTLS等gRPC参数。

PostgreSQL后端

pg后端使用PostgreSQL数据库，适合高负载场景（若负载过高可能需调整任务粒度）。依赖SELECT ... FOR UPDATE SKIP LOCKED查询实现高效任务声明。

示例代码：

go
package main

import (
  "context"

  "entrogo.com/entroq"
  "entrogo.com/entroq/backend/eqpg"
)

func main() {
  ctx := context.Background()
  eq := entroq.New(ctx, eqpg.Opener(":5432", eqpg.WithDB("postgres"), eqpg.WithUsername("myuser")))
  // 支持其他PostgreSQL相关参数

  // 使用eq.Modify, eq.Insert, eq.Claim等方法（通常在goroutine中）
}

注意：MySQL可实现类似后端，CockroachDB因不支持必要SQL语句无法替代PostgreSQL。

启动PostgreSQL实例

可通过Docker Compose在同一网络启动PostgreSQL和EntroQ服务。EntroQ会自动创建tasks表并处理版本更新。

docker-compose示例（生产环境不建议使用/tmp）：

yaml
version: "3"
services:
  database:
    image: "postgres:12"
    deploy:
      restart_policy:
        condition: any
    restart: always
    volumes:
      - /tmp/postgres/data:/var/lib/postgresql/data

  queue:
    image: "shiblon/entroq:v0.7"
    depends_on:
      - database
    deploy:
      restart_policy:
        condition: any
    restart: always
    ports:
      - 37706:37706
    command:
      - "pg"
      - "--dbaddr=database:5432"
      - "--attempts=10"

通过docker exec使用容器内eqc工具：

bash
docker exec shiblon/entroq:v0.7 eqc --help

QSvc

qsvc目录包含Docker镜像shiblon/entroq中的gRPC服务实现，暴露proto中定义的端点。可构建并启动不同后端的服务（如cmd/eqmemsvc、cmd/eqpgsvc），通过标志配置端口和后端连接。

服务应作为单例部署，客户端通过grpc后端连接。单例虽影响性能，但足以处理数千工作器，高负载通常源于任务粒度问题。

Python实现

contrib/py提供Python gRPC客户端及基础CLI，可替代Celery。通过pip安装：

bash
python3 -m pip install git+https://github.com/shiblon/entroq

模块包含协议缓冲区，使用示例见__main__.py中的命令行客户端实现。

示例

contrib/mr提供基于MapReduce的Go示例（压力测试用），展示竞争队列的典型用法：工作器、队列空检测、特定队列分配等。

授权

EntroQ本身不处理授权，直接访问后端时无授权机制。通过gRPC服务可集成授权，提供基于OPA的实现（支持内存和PostgreSQL后端）。

启用OPA-HTTP授权策略：服务命令行添加--authz=opahttp标志，需运行OPA实例并配置策略。

工作原理

eqc客户端可传递授权令牌，通过gRPC的Authorization: Bearer <token>头传输。服务将令牌及操作请求（如声明队列）转发至OPA，OPA返回授权结果。

OPA需包含entroq.authz包，结构符合authz.AuthzError类型（定义于pkg/authz/authz.go）。pkg/authz/opadata目录提供配置示例。