通知系统设计：push、email、SMS 重试和优先级

1. 概述

几乎每个产品都需要发送通知：OTP码通过SMS、订单确认邮件、新消息推送等。发送一条通知很容易。难的是发送数百万条通知而不丢失、不重复，且OTP不会排在营销邮件后面。

在本文中，我们从头开始设计一个多渠道通知系统。范围包括需求、数据模型、API、处理流程以及解决难点的方法：重复发送、DB与Kafka一致性、突发负载。示例代码使用Java 21、Spring Boot 3.x、PostgreSQL、Kafka和Redis。

2. 需求分析

2.1. 功能需求

• 其他服务（Order、Auth、Marketing等）调用API通过一个或多个渠道向用户发送通知：PUSH、EMAIL、SMS。
• 每条通知都有优先级：HIGH（OTP、安全警报）、NORMAL（交易）、LOW（营销）。
• 当提供商出现临时错误时，系统自动重试。达到最大次数后，通知被标记为FAILED供查询。
• 系统尊重用户偏好，例如用户已关闭营销邮件。

2.2. 规模和约束

• 每天约5000万条通知。平均近600条/秒，运行营销活动时峰值可达20000条/秒。
• HIGH优先级通知必须在几秒内发送，即使在峰值负载期间也是如此。
• SMS/Email提供商有速率限制，例如每个API密钥每秒500个请求。
• 基础设施层的传递语义是at-least-once，但必须进行合理的重复消除。用户收到两个OTP是糟糕的体验。收到两条需计费的SMS还要花钱。

3. 架构概览

Client service ──POST──▶ Notification API ──(1 transaction)──▶ PostgreSQL
                                                         │ notification, delivery (PENDING)
                                                         ▼
                                              Outbox Relay (poll, SKIP LOCKED)
                                                         │ publish 按优先级
                         ┌───────────────────────────────┼───────────────────────────┐
                         ▼                               ▼                           ▼
              notifications.high              notifications.normal          notifications.low
                         │                               │                           │
                         └────────────▶ Delivery Workers ◀───────────────────────────┘
                                         │  防重复(Redis) + 限速(Redis)
                                         ▼
                              FCM/APNs · Email provider · SMS provider
                                         │
                       成功 → SENT │ 临时错误 → PENDING + next_attempt_at

核心思想是：delivery表是唯一的真实源。 Kafka只是传输通道。重试无需单独的retry topic。Worker只需将记录设置为PENDING并设置未来的next_attempt_at，Outbox Relay到时自动重新推送。

4. 数据模型

CREATE TABLE notification (
    id              UUID PRIMARY KEY,
    idempotency_key VARCHAR(100) NOT NULL UNIQUE,
    user_id         BIGINT       NOT NULL,
    template        VARCHAR(50)  NOT NULL,
    payload         JSONB        NOT NULL,
    created_at      TIMESTAMPTZ  NOT NULL DEFAULT now()
);

CREATE TABLE delivery (
    id              UUID PRIMARY KEY,
    notification_id UUID        NOT NULL REFERENCES notification(id),
    channel         VARCHAR(10) NOT NULL,          -- PUSH | EMAIL | SMS
    priority        SMALLINT    NOT NULL,          -- 0 HIGH, 1 NORMAL, 2 LOW
    status          VARCHAR(10) NOT NULL,          -- PENDING | QUEUED | SENT | FAILED
    attempts        INT         NOT NULL DEFAULT 0,
    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    last_error      TEXT,
    UNIQUE (notification_id, channel)
);

CREATE INDEX idx_delivery_due
    ON delivery (priority, next_attempt_at)
    WHERE status = 'PENDING';

一些值得注意的要点：

• idempotency_key UNIQUE阻止客户端重复调用API创建相同通知。
• 每个渠道都是一个单独的delivery行。邮件失败不会导致push重新发送。
• priority在delivery中重复存储，以便partial index idx_delivery_due服务relay的确切查询。该索引只包含PENDING行，所以即使表有数十亿SENT行，索引仍然很小。

5. 通知接收API

首先，声明所需的依赖项：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <scope>runtime</scope>
</dependency>

接下来是API。客户端必须发送Idempotency-Key头。通常这是业务ID，例如order-123-confirmed。

public enum Priority { HIGH, NORMAL, LOW }
public enum Channel { PUSH, EMAIL, SMS }

public record SendRequest(long userId, String template, Map<String, Object> payload,
                          Priority priority, Set<Channel> channels) {}

@RestController
@RequestMapping("/api/notifications")
class NotificationController {

    private final NotificationService service;

    NotificationController(NotificationService service) { this.service = service; }

    @PostMapping
    ResponseEntity<Map<String, UUID>> send(@RequestHeader("Idempotency-Key") String key,
                                           @RequestBody SendRequest req) {
        UUID id = service.accept(key, req);
        return ResponseEntity.accepted().body(Map.of("id", id));
    }
}

API返回202 Accepted因为发送是异步进行的。主要处理逻辑在service中：

@Service
class NotificationService {

    private final JdbcTemplate jdbc;
    private final ObjectMapper mapper;

    NotificationService(JdbcTemplate jdbc, ObjectMapper mapper) {
        this.jdbc = jdbc;
        this.mapper = mapper;
    }

    @Transactional
    public UUID accept(String key, SendRequest req) {
        List<UUID> inserted = jdbc.queryForList("""
            INSERT INTO notification (id, idempotency_key, user_id, template, payload)
            VALUES (?, ?, ?, ?, ?::jsonb)
            ON CONFLICT (idempotency_key) DO NOTHING
            RETURNING id
            """, UUID.class, UUID.randomUUID(), key, req.userId(), req.template(), toJson(req.payload()));

        if (inserted.isEmpty()) {   // 重复请求 → 返回旧通知
            return jdbc.queryForObject(
                "SELECT id FROM notification WHERE idempotency_key = ?", UUID.class, key);
        }
        UUID id = inserted.getFirst();
        for (Channel ch : req.channels()) {
            jdbc.update("""
                INSERT INTO delivery (id, notification_id, channel, priority, status)
                VALUES (?, ?, ?, ?, 'PENDING')
                """, UUID.randomUUID(), id, ch.name(), req.priority().ordinal());
        }
        return id;
    }

    private String toJson(Map<String, Object> payload) {
        try {
            return mapper.writeValueAsString(payload);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("payload无效", e);
        }
    }
}

ON CONFLICT ... DO NOTHING RETURNING id在数据库中处理冲突。当两个相同key的请求同时到达时，只有一个请求插入成功。另一个请求收到空列表并重新读取旧ID。您不需要在应用层加锁。

注意，service在这个事务中不发送Kafka。这正是Transactional Outbox模式，下一部分会解释原因。

6. Outbox Relay和优先级队列

假设您先写DB然后立即发送Kafka。如果应用在两步之间崩溃，通知会留在DB中但永远不会被发送。反过来做，先发Kafka再写DB，那么worker可能处理一条不存在的记录。Outbox解决了这个问题：DB是transaction中唯一被写入的地方，relay则从DB读取并在之后发布。

@Component
class OutboxRelay {

    private static final String[] TOPICS =
        {"notifications.high", "notifications.normal", "notifications.low"};

    private final JdbcTemplate jdbc;
    private final KafkaTemplate<String, String> kafka;

    OutboxRelay(JdbcTemplate jdbc, KafkaTemplate<String, String> kafka) {
        this.jdbc = jdbc;
        this.kafka = kafka;
    }

    @Scheduled(fixedDelay = 200)
    @Transactional
    public void relay() {
        List<Map<String, Object>> due = jdbc.queryForList("""
            SELECT id, priority FROM delivery
            WHERE status = 'PENDING' AND next_attempt_at <= now()
            ORDER BY priority, next_attempt_at
            LIMIT 500
            FOR UPDATE SKIP LOCKED
            """);
        for (Map<String, Object> row : due) {
            UUID id = (UUID) row.get("id");
            int priority = ((Number) row.get("priority")).intValue();
            kafka.send(TOPICS[priority], id.toString(), id.toString()).join();
            jdbc.update("UPDATE delivery SET status = 'QUEUED' WHERE id = ?", id);
        }
    }
}

上面的代码做了三件事：

• ORDER BY priority, next_attempt_at优先获取HIGH行。即使有500万封营销邮件在等待，新的OTP仍然会在下一次poll时被选中。
• FOR UPDATE SKIP LOCKED允许多个relay实例并行运行。每个实例锁定不同的批次并跳过其他实例持有的行。没有实例需要等待另一个。
• .join()等待broker确认然后改变状态。如果relay在发送后但提交前崩溃，该行仍然是PENDING并会被重新发送。这是at-least-once。Worker必须防重，见7.2。

优先级在消费者端通过单独的topic和单独的concurrency再次得到保证：

@Component
class DeliveryListeners {

    private final DeliveryWorker worker;

    DeliveryListeners(DeliveryWorker worker) { this.worker = worker; }

    @KafkaListener(topics = "notifications.high", groupId = "delivery", concurrency = "12")
    void high(String deliveryId) { worker.process(UUID.fromString(deliveryId)); }

    @KafkaListener(topics = "notifications.normal", groupId = "delivery", concurrency = "6")
    void normal(String deliveryId) { worker.process(UUID.fromString(deliveryId)); }

    @KafkaListener(topics = "notifications.low", groupId = "delivery", concurrency = "2")
    void low(String deliveryId) { worker.process(UUID.fromString(deliveryId)); }
}

Kafka在单个topic中没有优先级队列。因此，分离topic是最实际的方法。当营销活动使notifications.low堆积数百万条消息时，notifications.high的消费者仍然在自己的线程上运行且不受影响。concurrency数字必须小于或等于相应topic的分区数。

7. Worker发送和重试

7.1. 重试策略

不是所有错误都应该重试。超时或HTTP 429/503是临时错误。错误的电话号码或被撤销的push token是永久错误，重试只是浪费钱。

public class RetryPolicy {

    private final Duration base;
    private final Duration max;
    private final int maxAttempts;

    public RetryPolicy(Duration base, Duration max, int maxAttempts) {
        this.base = base;
        this.max = max;
        this.maxAttempts = maxAttempts;
    }

    /** attempt: 已尝试的次数（从1开始）。Empty = 停止重试。 */
    public Optional<Duration> nextDelay(int attempt) {
        if (attempt >= maxAttempts) return Optional.empty();
        long millis = base.toMillis() * (1L << Math.min(attempt - 1, 20));
        return Optional.of(Duration.ofMillis(Math.min(millis, max.toMillis())));
    }
}

使用base = 2s、max = 5m、maxAttempts = 6，等待间隔依次为：

第1次失败 → 等待 2s
第2次失败 → 等待 4s
第3次失败 → 等待 8s
第4次失败 → 等待 16s
第5次失败 → 等待 32s
第6次失败 → FAILED

在生产环境中，您应该在等待间隔中添加随机jitter。否则，当提供商故障时，数千条有相同错误的通知会在完全相同的时刻重试。这种现象称为thundering herd。

您可以用一个简单的测试来验证这个策略：

class RetryPolicyTest {

    private final RetryPolicy policy =
        new RetryPolicy(Duration.ofSeconds(2), Duration.ofSeconds(10), 6);

    @Test
    void 退避按指数增长且受上限限制() {
        assertThat(policy.nextDelay(1)).contains(Duration.ofSeconds(2));
        assertThat(policy.nextDelay(3)).contains(Duration.ofSeconds(8));
        assertThat(policy.nextDelay(4)).contains(Duration.ofSeconds(10)); // 受max限制
    }

    @Test
    void 达到最大次数时停止重试() {
        assertThat(policy.nextDelay(6)).isEmpty();
    }
}

7.2. Worker：防重复、限速、更新状态

public class TransientSendException extends RuntimeException {
    public TransientSendException(String msg) { super(msg); }
}

public interface ChannelSender {
    Channel channel();
    void send(UUID deliveryId);   // 如果是临时错误则抛出TransientSendException
}

@Service
class DeliveryWorker {

    private final JdbcTemplate jdbc;
    private final StringRedisTemplate redis;
    private final Map<Channel, ChannelSender> senders;
    private final RetryPolicy retry = new RetryPolicy(Duration.ofSeconds(2), Duration.ofMinutes(5), 6);

    DeliveryWorker(JdbcTemplate jdbc, StringRedisTemplate redis, List<ChannelSender> list) {
        this.jdbc = jdbc;
        this.redis = redis;
        this.senders = list.stream().collect(Collectors.toMap(ChannelSender::channel, s -> s));
    }

    public void process(UUID id) {
        Map<String, Object> d = jdbc.queryForMap(
            "SELECT channel, attempts, status FROM delivery WHERE id = ?", id);
        if ("SENT".equals(d.get("status"))) return;               // 来自Kafka的重复message

        Boolean first = redis.opsForValue()
            .setIfAbsent("notif:sending:" + id, "1", Duration.ofMinutes(10));
        if (!Boolean.TRUE.equals(first)) return;                  // 另一个worker正在发送

        Channel channel = Channel.valueOf((String) d.get("channel"));
        int attempt = ((Number) d.get("attempts")).intValue() + 1;
        try {
            if (!acquireRate(channel)) {
                throw new TransientSendException("rate limited");  // 延后，不计为provider错误
            }
            senders.get(channel).send(id);
            jdbc.update("UPDATE delivery SET status='SENT', attempts=? WHERE id=?", attempt, id);
        } catch (TransientSendException e) {
            redis.delete("notif:sending:" + id);
            retry.nextDelay(attempt).ifPresentOrElse(
                delay -> jdbc.update("""
                    UPDATE delivery SET status='PENDING', attempts=?, last_error=?,
                           next_attempt_at = now() + make_interval(secs => ?)
                    WHERE id=?""", attempt, e.getMessage(), delay.toSeconds(), id),
                () -> markFailed(id, attempt, e.getMessage()));
        } catch (RuntimeException e) {                             // 永久错误
            markFailed(id, attempt, e.getMessage());
        }
    }

    private void markFailed(UUID id, int attempt, String error) {
        jdbc.update("UPDATE delivery SET status='FAILED', attempts=?, last_error=? WHERE id=?",
            attempt, error, id);
    }

    private boolean acquireRate(Channel channel) {
        String key = "rate:" + channel + ":" + Instant.now().getEpochSecond();
        Long count = redis.opsForValue().increment(key);
        if (count != null && count == 1) redis.expire(key, Duration.ofSeconds(2));
        return count != null && count <= 500;
    }
}

Worker有两层防重复机制：

1. **检查status = 'SENT'**排除已发送后再次发布的消息。
2. 带TTL的Redis SET NX锁阻止两个消费者收到同一条消息的情况，例如Kafka中途rebalance时。10分钟的TTL确保如果worker崩溃，锁会自动释放。

这仍然不是exactly-once。如果worker成功调用提供商但在UPDATE ... 'SENT'之前崩溃，通知可能在锁过期后被重新发送。要更严格，可以将deliveryId作为幂等性key传给提供商（如果提供商支持）。许多电子邮件提供商和FCM都有这种机制。

acquireRate是Redis上的固定窗口速率限制器。所有worker实例为每个渠道共享每秒计数器，因此总速率不会超过提供商的500请求/秒配额。超过阈值时，通知被推回PENDING而不是调用提供商并收到HTTP 429。

8. 难点和解决方案

8.1. 突发负载

当Marketing一次发送1000万条通知时，系统在各个层面都会承受负载：

• API只是插入DB然后返回，所以能承受高负载。对于大型活动，您应该提供单独的批量API来接收用户列表或段。后台job将按1000行批次生成delivery行。
• Kafka吸收剩余部分。low topic的积压是正常的，只要high topic仍然在运转。
• 速率限制器保护提供商。LOW通知发送得慢一点也没人抱怨。

应该添加一个额外的措施：在acquireRate中，为HIGH预留一部分配额，例如LOW只能使用60%的阈值。这样，OTP永远不会因为营销邮件而被限速。

8.2. DB与Kafka一致性

Outbox结合at-least-once和幂等consumer是标准的三层保证。您不需要分布式事务（XA）。代价是增加一个poll周期的延迟，这里是200毫秒。如果需要更低的延迟，可以用CDC（读取PostgreSQL的WAL）替代polling，但基础设施会更复杂。

8.3. 数据膨胀

delivery表每天增加约6000-8000万行。您应该按日期对表进行分区（PARTITION BY RANGE (created_at)）并删除超过30-90天的分区。第4部分的partial index使relay查询始终保持快速，但无法替代数据清理。

8.4. 用户偏好和安静时段

在NotificationService中创建delivery时立即检查偏好设置。用户已关闭的渠道不创建行。对于静默时段，例如晚上22点后不发送营销push，只需将next_attempt_at设置为第二天早上8点。relay机制会处理其余部分，无需额外的scheduler。

**注意：**不要在长@Transactional内调用提供商。一个慢30秒的HTTP请求会持有DB连接和行锁，连接池会很快耗尽。在DeliveryWorker中，每个UPDATE是一个短的、与提供商调用无关的独立事务。

**注意：**不要无限重试，也不要重试永久错误。一个错误的电话号码，重试6次，乘以100万用户是一笔巨大的SMS账单。务必仔细分类每个提供商的错误代码。

8.5. 何时不应该使用这个设计

如果您每天只发送几千封邮件，PostgreSQL加SKIP LOCKED就够了。Worker直接poll delivery表，完全跳过Kafka。Kafka只有在需要大规模按优先级拆分消费者或多个系统需要消费同一流事件时才值得。反之，如果需要保证同一用户的通知间的绝对顺序，上面的设计还不够。您需要使用userId作为Kafka key并跳过这些通知的异步重试。

9. 结论

我们设计了一个多渠道通知系统，主要决策如下：

• delivery表是唯一的真实源。 重试只是重新设置PENDING并设置next_attempt_at。
• 带FOR UPDATE SKIP LOCKED的Transactional Outbox在不需要分布式事务的情况下保持DB和Kafka一致，同时允许多个relay并行运行。
• 优先级在两层得到保证： SQL中的poll顺序，以及Kafka中单独的topic和concurrency。
• 多层防重复： API中的Idempotency-Key、worker中的状态检查和Redis锁、发送给提供商的幂等性key。
• Redis上的共享速率限制器在突发负载时保护提供商的配额。

从这个框架，您可以进一步扩展多语言模板、从提供商接收“已交付/已读”状态的webhook，或按渠道监控FAILED比率的仪表板。