会员日前的一份八百万名单
双十一前的会员日，运营在群里发来一份圈选条件：近 30 天活跃过、带 vip 或 trial 标签、排除已退订和黑名单用户，预计八百万（案例假设值）。要求当晚 20:00 全量推送，理由是“会员日只有一次，早发早转化”。
工程同学把条件直接拼成一条 SQL，导出一份 ID 文件，交给推送任务逐条发送。任务启动十分钟后，推送厂商的限流开始返回 429，队列迅速堆积；二十分钟后，订单状态变更通知也排在同一队列后面，用户付款后半小时没收到确认。真正的事故不是推送没发出去，而是营销流量挤占了事务通知。
这个场景里有三个决定：圈选规则是否可解释、触达是否分批、营销与事务是否隔离。本文只讨论前两个，第三个放在 9.3。
圈选条件要能逐条解释
圈选不是“查出一批 ID”这么简单。运营给的每个自然语言条件，都要能映射成服务端可复现的字段和比较方式，否则活动结束后没人能回答“为什么这个人收到了”。
| 运营条件 | 服务端表达 | 需要说清楚的点 |
|---|---|---|
| 近 30 天活跃 | LastActive >= now-30d | 边界是否包含当天；时区以哪个为准 |
| 会员或试用 | Tags 任一命中 | 是“任一”不是“全部”；空标签是否放行 |
| 已退订、黑名单 | Exclude[userID] | 排除优先于其他条件，不能被标签覆盖 |
| 用户 ID 为空 | 跳过 | 脏数据不能进入发送队列 |
| 结果顺序 | 按 ID 升序 | 分批和重跑时顺序稳定，便于对账 |
最容易出错的是“任一命中”被写成“全部命中”。运营说“会员和试用用户”，口语里是两类人，代码里如果写成标签交集，人数会从八百万掉到几十万，而且不会报错。第二个坑是空标签：新用户还没有标签，如果规则要求“标签包含 vip”，他们会被静默排除；如果规则是“标签为空表示不限”，又可能把全量用户圈进来。这两种语义都必须在规则定义里写死，不能靠调用方猜。
排除名单的优先级也要固定。一个用户同时有 vip 标签和在退订表里，结果必须是排除。把排除放在最后做“二次过滤”看似等价，但一旦中间有分批、重试或人工补发，就容易绕过。Select 把排除放在标签判断之前，就是为了让“排除优先”成为代码结构，而不是口头约定。
失败场景：一次全量推送打满通道
回到会员日。八百万条推送如果按 20:00 一次性入队，假设推送厂商的单通道容量是每秒五万条（案例假设值），光入队就需要约 160 秒。问题在于入队不等于送达：每条消息还要渲染模板、查用户设备、做厂商路由，实际吞吐远低于入队速度。队列在半小时内堆到几百万条，重试又把失败消息重新放回队尾。
更糟的是，订单通知和营销推送共用了一个队列。营销消息的优先级低，但队列是先进先出，事务通知被压在后面。用户付款后收不到确认，客服工单在当晚涨了三倍。事后看，推送任务本身没有 bug，错在把“全量”当成了一个批次，并且没有把营销流量与事务流量分开。
分批不是简单地把八百万切成八个一百万。正确做法是按通道容量和下游承受能力算批次大小，批次之间留出间隔，并且给每批设置独立的超时和失败上限。第一批失败率超过阈值就暂停，而不是继续把后面七批推下去。触达节奏本质上是背压：上游生产速度不能长期高于下游消费速度。
Go 实现：Select 只做确定性过滤
Select 不负责分批、发送和重试，只负责把一份用户快照按规则过滤成确定的结果。它不修改入参，结果按 ID 排序，方便调用方按固定顺序切批和重跑。
// file: internal/reach/audience.go
// tier: L1
// verify: go test ./internal/reach/
package reach
import (
"sort"
"strings"
)
// User 是一条参与圈选的用户记录。Tags 是运营维护的标签，例如 vip、
// trial、churn_risk；LastActive 是服务端记录的最近活跃时间，单位由调用方
// 统一，通常是 Unix 秒。标签和活跃时间都来自离线或近线数仓，不是实时值。
type User struct {
ID string
Tags []string
LastActive int64
}
// Rule 描述一次圈选。Tags 之间是“任一命中”，不是“全部命中”；空 Tags
// 表示不限制标签。ActiveSince 为 0 表示不限制活跃时间，否则要求
// LastActive 大于等于该值。Exclude 是按用户 ID 索引的排除名单，优先于
// 其他条件。
type Rule struct {
Tags []string
ActiveSince int64
Exclude map[string]bool
}
// Select 返回命中规则的用户。结果按 ID 升序，输入切片和其中的元素都不会
// 被修改。返回的 User 是值拷贝，Tags 也做了复制，调用方修改结果不会影响
// 传入的数据。
//
// 过滤顺序固定为：空 ID、排除名单、活跃时间、标签。顺序本身不影响结果，
// 但固定顺序让日志和排查更容易复现。
func Select(users []User, r Rule) []User {
wanted := make(map[string]struct{}, len(r.Tags))
for _, tag := range r.Tags {
tag = strings.TrimSpace(tag)
if tag != "" {
wanted[tag] = struct{}{}
}
}
out := make([]User, 0, len(users))
for _, u := range users {
if u.ID == "" {
continue
}
if r.Exclude[u.ID] {
continue
}
if r.ActiveSince != 0 && u.LastActive < r.ActiveSince {
continue
}
if len(wanted) > 0 && !hasAnyTag(u.Tags, wanted) {
continue
}
u.Tags = append([]string(nil), u.Tags...)
out = append(out, u)
}
sort.SliceStable(out, func(i, j int) bool {
return out[i].ID < out[j].ID
})
return out
}
// hasAnyTag 判断用户标签是否至少命中一个圈选标签。空字符串在两边都被
// 忽略，避免脏数据把无关用户圈进来。
func hasAnyTag(tags []string, wanted map[string]struct{}) bool {
for _, tag := range tags {
if _, ok := wanted[strings.TrimSpace(tag)]; ok {
return true
}
}
return false
}
实现里有几个取舍：
- 标签先去空白再匹配，规则里的
" vip "和用户身上的"vip"视为同一个标签。脏数据不应该让圈选结果漂移。 - 空
Tags表示不限制标签，而不是“匹配不到任何人”。这个语义在代码注释和规则配置里都要写清楚。 ActiveSince=0表示不限制活跃时间。用 0 而不是负数做哨兵，是因为时间戳 0 本身没有业务含义。Exclude用 map，查找是 O(1)；nil map 也可以安全读取，调用方不需要先初始化。- 输出做了值拷贝和标签复制，调用方后续修改结果不会污染输入快照。
- 排序用
sort.SliceStable，相同 ID 的相对顺序保持稳定。正常数据里 ID 唯一，但脏数据可能重复，稳定排序让重跑结果一致。 - 函数不返回错误。圈选规则是纯过滤，没有 I/O，也不应该因为某个用户数据脏就整体失败；脏数据被跳过，由上游对账发现。
测试：任一命中、时间边界与排除优先
测试要覆盖四个容易回归的语义：标签任一命中、空标签放行、活跃时间边界、排除优先。另外还要验证排序和入参不被修改。
// file: internal/reach/audience_test.go
// tier: L1
// verify: go test ./internal/reach/
package reach
import (
"reflect"
"testing"
)
func TestSelectAnyTagMatches(t *testing.T) {
users := []User{
{ID: "u1", Tags: []string{"vip"}, LastActive: 100},
{ID: "u2", Tags: []string{"trial", "new"}, LastActive: 100},
{ID: "u3", Tags: []string{"churn_risk"}, LastActive: 100},
}
got := Select(users, Rule{Tags: []string{"vip", "trial"}})
want := []string{"u1", "u2"}
if ids := userIDs(got); !reflect.DeepEqual(ids, want) {
t.Fatalf("Select ids = %v; want %v", ids, want)
}
}
func TestSelectEmptyTagsMatchesAll(t *testing.T) {
users := []User{
{ID: "u2", Tags: []string{"trial"}, LastActive: 50},
{ID: "u1", Tags: nil, LastActive: 100},
}
got := Select(users, Rule{})
if ids := userIDs(got); !reflect.DeepEqual(ids, []string{"u1", "u2"}) {
t.Fatalf("Select ids = %v; want [u1 u2]", ids)
}
}
func TestSelectActiveSince(t *testing.T) {
users := []User{
{ID: "u1", LastActive: 99},
{ID: "u2", LastActive: 100},
{ID: "u3", LastActive: 101},
}
got := Select(users, Rule{ActiveSince: 100})
if ids := userIDs(got); !reflect.DeepEqual(ids, []string{"u2", "u3"}) {
t.Fatalf("Select ids = %v; want [u2 u3]", ids)
}
if ids := userIDs(Select(users, Rule{})); len(ids) != 3 {
t.Fatalf("ActiveSince=0 ids = %v; want all users", ids)
}
}
func TestSelectExcludeWins(t *testing.T) {
users := []User{
{ID: "u1", Tags: []string{"vip"}, LastActive: 100},
{ID: "u2", Tags: []string{"vip"}, LastActive: 100},
}
got := Select(users, Rule{
Tags: []string{"vip"},
ActiveSince: 100,
Exclude: map[string]bool{"u1": true},
})
if ids := userIDs(got); !reflect.DeepEqual(ids, []string{"u2"}) {
t.Fatalf("Select ids = %v; want [u2]", ids)
}
}
func TestSelectSortsAndKeepsInput(t *testing.T) {
users := []User{
{ID: "u3", Tags: []string{"vip"}, LastActive: 100},
{ID: "u1", Tags: []string{"vip"}, LastActive: 100},
{ID: "u2", Tags: []string{"vip"}, LastActive: 100},
}
snapshot := append([]User(nil), users...)
got := Select(users, Rule{Tags: []string{"vip"}})
if ids := userIDs(got); !reflect.DeepEqual(ids, []string{"u1", "u2", "u3"}) {
t.Fatalf("Select ids = %v; want sorted [u1 u2 u3]", ids)
}
if !reflect.DeepEqual(users, snapshot) {
t.Fatalf("Select mutated input: %v", users)
}
}
func TestSelectCopiesTagsAndDropsEmptyID(t *testing.T) {
users := []User{
{ID: "", Tags: []string{"vip"}, LastActive: 100},
{ID: "u1", Tags: []string{"vip"}, LastActive: 100},
}
got := Select(users, Rule{Tags: []string{" vip "}})
if len(got) != 1 || got[0].ID != "u1" {
t.Fatalf("Select = %+v; want only u1", got)
}
got[0].Tags[0] = "changed"
if users[1].Tags[0] != "vip" {
t.Fatalf("result tags alias input: %v", users[1].Tags)
}
}
func TestSelectNoMatchReturnsEmpty(t *testing.T) {
got := Select([]User{{ID: "u1", Tags: []string{"new"}}}, Rule{Tags: []string{"vip"}})
if got == nil || len(got) != 0 {
t.Fatalf("Select = %v; want non-nil empty slice", got)
}
}
func userIDs(users []User) []string {
ids := make([]string, len(users))
for i, u := range users {
ids[i] = u.ID
}
return ids
}
TestSelectAnyTagMatches 验证 vip 和 trial 是“任一命中”，不是交集。TestSelectEmptyTagsMatchesAll 固定空规则放行全部用户的语义。TestSelectActiveSince 把边界放在 LastActive == ActiveSince，确认是大于等于而不是大于；同时验证 ActiveSince=0 不限制。TestSelectExcludeWins 让被排除用户同时满足标签和活跃条件，确认排除优先。
TestSelectSortsAndKeepsInput 用乱序输入验证输出升序，并对比调用前后的输入快照。TestSelectCopiesTagsAndDropsEmptyID 验证空 ID 被丢弃，以及修改结果里的 Tags 不会影响入参。TestSelectNoMatchReturnsEmpty 确认无命中时返回非 nil 的空切片，避免调用方把 nil 当成错误。
运行命令与真实输出
在 webapp-lab 目录执行：
go test ./internal/reach/ -run '^TestSelect' -count=1 -v
一次本机运行的真实输出如下，耗时随机器变化：
=== RUN TestSelectAnyTagMatches
--- PASS: TestSelectAnyTagMatches (0.00s)
=== RUN TestSelectEmptyTagsMatchesAll
--- PASS: TestSelectEmptyTagsMatchesAll (0.00s)
=== RUN TestSelectActiveSince
--- PASS: TestSelectActiveSince (0.00s)
=== RUN TestSelectExcludeWins
--- PASS: TestSelectExcludeWins (0.00s)
=== RUN TestSelectSortsAndKeepsInput
--- PASS: TestSelectSortsAndKeepsInput (0.00s)
=== RUN TestSelectCopiesTagsAndDropsEmptyID
--- PASS: TestSelectCopiesTagsAndDropsEmptyID (0.00s)
=== RUN TestSelectNoMatchReturnsEmpty
--- PASS: TestSelectNoMatchReturnsEmpty (0.00s)
PASS
ok webapp/internal/reach 0.231s
七条用例全部通过。它们固定的是单次圈选的行为；分批大小、通道容量和队列隔离属于发送编排，需要在上层用配置和监控约束。
小结
圈选规则要能逐条翻译成字段和比较方式，标签是任一命中，排除优先，空条件要明确语义。触达要按下游容量分批，营销与事务分开，否则一次“全量”就会把正常通知挤到两小时后。
Select 只做确定性过滤：不改入参、结果排序、脏数据跳过。下一节讨论用户被触达之后，怎样判断回流到底来自哪一次活动。
阅读导航：上一节：目录与学习路径 · 下一节：9.2 召回链路与转化归因 。
常见错误与边界
- 把“会员和试用”写成标签交集。运营说的是两类人，代码却要求同时满足，人数骤降且不报错。
- 空标签被当成“匹配不到”。新用户没有标签，会被静默排除；空规则应明确为不限制。
- 排除名单最后才做。分批、重试、人工补发都可能绕过二次过滤，排除必须前置。
- 用
>而不是>=判断活跃时间。边界当天的用户会被漏掉，活动人数对不上。 - 直接排序入参。调用方后续还要用原始顺序对账，
Select应输出新切片。 - 全量一次入队。下游吞吐低于入队速度时，队列会堆到不可控，重试进一步放大。
- 营销与事务共用队列。营销峰值会延迟订单通知，事故影响从“推送慢”升级为“交易不可用”。
- 只按人数分批，不按通道容量分批。八百万切八批仍然可能每批都超过下游承受能力。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。