Thiết kế hệ thống thông báo: push, email, SMS với retry và ưu tiên

1. Tổng quan

Gần như mọi sản phẩm đều cần gửi thông báo: mã OTP qua SMS, email xác nhận đơn hàng, push khi có tin nhắn mới. Gửi một thông báo thì dễ. Khó là gửi hàng triệu thông báo mà không mất, không trùng, và OTP không phải xếp hàng sau email khuyến mãi.

Trong bài này, chúng ta thiết kế hệ thống thông báo đa kênh từ đầu đến cuối. Phạm vi gồm yêu cầu, mô hình dữ liệu, API, luồng xử lý và cách giải các điểm khó: gửi trùng, nhất quán giữa DB và Kafka, tải đột biến. Code minh hoạ dùng Java 21, Spring Boot 3.x, PostgreSQL, Kafka và Redis.

2. Yêu cầu bài toán

2.1. Yêu cầu chức năng

• Các service khác (Order, Auth, Marketing…) gọi API để gửi thông báo cho một người dùng qua một hoặc nhiều kênh: PUSH, EMAIL, SMS.
• Mỗi thông báo có độ ưu tiên: HIGH (OTP, cảnh báo bảo mật), NORMAL (giao dịch), LOW (marketing).
• Hệ thống tự retry khi nhà cung cấp lỗi tạm thời. Sau số lần tối đa, thông báo bị đánh dấu FAILED để tra cứu.
• Hệ thống tôn trọng tuỳ chọn người dùng, ví dụ người dùng đã tắt email marketing.

2.2. Quy mô và ràng buộc

• Khoảng 50 triệu thông báo/ngày. Trung bình gần 600/giây, đỉnh khi chạy chiến dịch marketing có thể lên 20.000/giây.
• Thông báo HIGH phải gửi trong vòng vài giây, kể cả lúc đỉnh tải.
• Nhà cung cấp SMS/email giới hạn tốc độ, ví dụ 500 request/giây cho mỗi API key.
• Ngữ nghĩa giao nhận là at-least-once ở tầng hạ tầng, nhưng phải chống trùng ở mức hợp lý. Người dùng nhận hai mã OTP là trải nghiệm tệ. Nhận hai SMS tính phí còn tốn tiền.

3. Kiến trúc tổng thể

Client service ──POST──▶ Notification API ──(1 transaction)──▶ PostgreSQL
                                                         │ notification, delivery (PENDING)
                                                         ▼
                                              Outbox Relay (poll, SKIP LOCKED)
                                                         │ publish theo priority
                         ┌───────────────────────────────┼───────────────────────────┐
                         ▼                               ▼                           ▼
              notifications.high              notifications.normal          notifications.low
                         │                               │                           │
                         └────────────▶ Delivery Workers ◀───────────────────────────┘
                                         │  dedup (Redis) + rate limit (Redis)
                                         ▼
                              FCM/APNs · Email provider · SMS provider
                                         │
                       thành công → SENT │ lỗi tạm thời → PENDING + next_attempt_at

Ý tưởng chính: bảng delivery là nguồn sự thật. Kafka chỉ là đường vận chuyển. Retry không cần retry topic riêng. Worker chỉ cần đặt bản ghi về PENDING với next_attempt_at ở tương lai, Outbox Relay sẽ tự đẩy lại khi đến hạn.

4. Mô hình dữ liệu

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';

Một số điểm đáng chú ý:

• idempotency_key UNIQUE chặn việc client gọi lại API tạo thông báo trùng.
• Mỗi kênh là một dòng delivery riêng. Email lỗi không làm push phải gửi lại.
• priority được lặp lại trong delivery để partial index idx_delivery_due phục vụ đúng câu truy vấn của relay. Index chỉ chứa các dòng PENDING nên luôn nhỏ, dù bảng có hàng tỷ dòng SENT.

5. API nhận thông báo

Đầu tiên, bạn khai báo các dependency cần dùng:

<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>

Tiếp theo là API. Client bắt buộc gửi header Idempotency-Key. Thường đó là ID nghiệp vụ, ví dụ 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 trả 202 Accepted vì việc gửi diễn ra bất đồng bộ. Phần xử lý chính nằm trong 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()) {   // request lặp lại → trả về thông báo cũ
            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 không hợp lệ", e);
        }
    }
}

ON CONFLICT ... DO NOTHING RETURNING id xử lý tranh chấp ngay trong database. Khi hai request cùng key đến đồng thời, chỉ một request insert thành công. Request còn lại nhận về danh sách rỗng và đọc lại ID cũ. Bạn không cần lock ở tầng ứng dụng.

Lưu ý rằng service không gửi Kafka trong transaction này. Đó chính là pattern Transactional Outbox, phần tiếp theo sẽ giải thích lý do.

6. Outbox Relay và hàng đợi theo độ ưu tiên

Giả sử bạn ghi DB rồi gửi Kafka ngay sau đó. Nếu ứng dụng chết giữa hai bước, thông báo nằm trong DB mà không bao giờ được gửi. Làm ngược lại, gửi Kafka trước rồi mới ghi DB, thì worker có thể xử lý một bản ghi chưa tồn tại. Outbox giải quyết vấn đề này: DB là nơi duy nhất được ghi trong transaction, còn relay đọc DB và publish sau.

@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);
        }
    }
}

Code trên làm ba việc:

• ORDER BY priority, next_attempt_at lấy các dòng HIGH trước. Dù có 5 triệu email marketing đang chờ, OTP mới vẫn được chọn ở lần poll kế tiếp.
• FOR UPDATE SKIP LOCKED cho phép chạy nhiều instance relay song song. Mỗi instance khóa một lô khác nhau và bỏ qua các dòng instance khác đang giữ. Không instance nào phải chờ instance nào.
• .join() chờ broker xác nhận rồi mới đổi trạng thái. Nếu relay chết sau khi gửi nhưng trước khi commit, dòng vẫn ở PENDING và sẽ được gửi lại. Đây là at-least-once. Worker phải chống trùng, xem mục 7.2.

Độ ưu tiên được đảm bảo lần nữa ở phía consumer qua topic riêng và concurrency riêng:

@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 không có priority queue trong một topic. Vì vậy, tách topic là cách thực tế nhất. Khi chiến dịch marketing làm notifications.low tồn hàng triệu message, consumer của notifications.high vẫn chạy trên thread riêng và không bị ảnh hưởng. Số concurrency phải nhỏ hơn hoặc bằng số partition của topic tương ứng.

7. Worker gửi và retry

7.1. Chính sách retry

Không phải lỗi nào cũng nên retry. Timeout hoặc HTTP 429/503 là lỗi tạm thời. Số điện thoại sai hay token push đã bị thu hồi là lỗi vĩnh viễn, retry chỉ tốn tiền.

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: số lần đã thử (bắt đầu từ 1). Empty = dừng retry. */
    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())));
    }
}

Với base = 2s, max = 5m, maxAttempts = 6, các khoảng chờ lần lượt là:

lần 1 lỗi → chờ 2s
lần 2 lỗi → chờ 4s
lần 3 lỗi → chờ 8s
lần 4 lỗi → chờ 16s
lần 5 lỗi → chờ 32s
lần 6 lỗi → FAILED

Trong production, bạn nên cộng thêm jitter ngẫu nhiên vào khoảng chờ. Nếu không, hàng nghìn thông báo cùng lỗi lúc provider sập sẽ retry đúng cùng một thời điểm. Hiện tượng này gọi là thundering herd.

Bạn có thể kiểm tra chính sách này bằng một test đơn giản:

class RetryPolicyTest {

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

    @Test
    void backoffTangTheoLuyThuaVaBiChanTren() {
        assertThat(policy.nextDelay(1)).contains(Duration.ofSeconds(2));
        assertThat(policy.nextDelay(3)).contains(Duration.ofSeconds(8));
        assertThat(policy.nextDelay(4)).contains(Duration.ofSeconds(10)); // bị chặn bởi max
    }

    @Test
    void dungRetryKhiDatSoLanToiDa() {
        assertThat(policy.nextDelay(6)).isEmpty();
    }
}

7.2. Worker: chống trùng, giới hạn tốc độ, cập nhật trạng thái

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

public interface ChannelSender {
    Channel channel();
    void send(UUID deliveryId);   // ném TransientSendException nếu lỗi tạm thời
}

@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;               // message lặp từ Kafka

        Boolean first = redis.opsForValue()
            .setIfAbsent("notif:sending:" + id, "1", Duration.ofMinutes(10));
        if (!Boolean.TRUE.equals(first)) return;                  // worker khác đang gửi

        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");  // đẩy lùi, không tính là lỗi 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) {                             // lỗi vĩnh viễn
            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 có hai lớp chống trùng:

1. Kiểm tra status = 'SENT' loại bỏ message bị publish lại sau khi đã gửi xong.
2. Khóa Redis SET NX có TTL chặn trường hợp hai consumer cùng nhận một message, ví dụ khi Kafka rebalance giữa chừng. TTL 10 phút đảm bảo khóa tự nhả nếu worker chết.

Đây vẫn chưa phải exactly-once. Nếu worker gọi provider thành công rồi chết trước khi UPDATE ... 'SENT', thông báo có thể bị gửi lại sau khi khóa hết hạn. Muốn chặt hơn, bạn truyền deliveryId làm idempotency key sang provider, nếu provider hỗ trợ. Nhiều email provider và FCM có cơ chế này.

acquireRate là rate limiter dạng fixed window trên Redis. Mọi instance worker dùng chung một bộ đếm mỗi giây cho từng kênh, nên tổng tốc độ không vượt quota 500 request/giây của provider. Khi vượt ngưỡng, thông báo được đẩy lùi về PENDING thay vì gọi provider và nhận HTTP 429.

8. Các điểm khó và cách giải

8.1. Tải đột biến

Khi Marketing gửi 10 triệu thông báo một lúc, hệ thống chịu tải theo từng tầng:

• API chỉ insert DB rồi trả về, nên chịu được tải cao. Với chiến dịch lớn, bạn nên cung cấp một API batch riêng nhận danh sách user hoặc một segment. Một background job sẽ sinh các dòng delivery theo lô 1.000 dòng.
• Kafka hấp thụ phần dư. Topic low tồn đọng là chuyện bình thường, miễn topic high vẫn thông.
• Rate limiter bảo vệ provider. Thông báo LOW gửi chậm hơn một chút cũng không ai phàn nàn.

Nên thêm một biện pháp nữa: trong acquireRate, dành riêng một phần quota cho HIGH, ví dụ LOW chỉ được dùng 60% ngưỡng. Như vậy, OTP không bao giờ bị rate limit vì email khuyến mãi.

8.2. Nhất quán giữa DB và Kafka

Outbox kết hợp at-least-once và consumer idempotent là bộ ba tiêu chuẩn. Bạn không cần distributed transaction (XA). Cái giá phải trả là độ trễ thêm khoảng một chu kỳ poll, ở đây là 200ms. Nếu cần độ trễ thấp hơn, bạn thay polling bằng CDC (đọc WAL của PostgreSQL), nhưng hạ tầng sẽ phức tạp hơn.

8.3. Dữ liệu phình to

Mỗi ngày bảng delivery có thêm khoảng 60–80 triệu dòng. Bạn nên partition bảng theo ngày (PARTITION BY RANGE (created_at)) và drop các partition cũ hơn 30–90 ngày. Partial index ở mục 4 giữ cho truy vấn của relay luôn nhanh, nhưng không thay thế được việc dọn dữ liệu.

8.4. Tuỳ chọn người dùng và giờ yên lặng

Hãy kiểm tra tuỳ chọn ngay lúc tạo delivery trong NotificationService. Kênh nào người dùng đã tắt thì không tạo dòng. Với quiet hours, ví dụ không gửi push marketing sau 22 giờ, bạn chỉ cần đặt next_attempt_at bằng 8 giờ sáng hôm sau. Cơ chế relay xử lý phần còn lại, không cần thêm scheduler nào.

Lưu ý: Đừng gọi provider bên trong @Transactional dài. Một request HTTP chậm 30 giây sẽ giữ DB connection và row lock suốt thời gian đó, và pool cạn rất nhanh. Trong DeliveryWorker, mỗi UPDATE là một transaction ngắn, độc lập với lời gọi provider.

Lưu ý: Đừng retry vô hạn và đừng retry lỗi vĩnh viễn. Một số điện thoại sai, retry 6 lần, nhân với 1 triệu người dùng là hoá đơn SMS rất lớn. Hãy phân loại mã lỗi của từng provider thật cẩn thận.

8.5. Khi nào KHÔNG nên dùng thiết kế này

Nếu bạn chỉ gửi vài nghìn email/ngày, PostgreSQL cộng SKIP LOCKED là đủ. Worker poll trực tiếp bảng delivery, bỏ hẳn Kafka. Kafka chỉ đáng giá khi bạn cần tách consumer theo priority ở quy mô lớn, hoặc khi nhiều hệ thống khác cần tiêu thụ cùng luồng sự kiện. Ngược lại, nếu bạn cần đảm bảo thứ tự tuyệt đối giữa các thông báo của cùng một người dùng, thiết kế trên chưa đủ. Bạn phải dùng userId làm Kafka key và bỏ retry bất đồng bộ cho các thông báo đó.

9. Kết luận

Chúng ta đã thiết kế một hệ thống thông báo đa kênh với các quyết định chính sau:

• Bảng delivery là nguồn sự thật. Retry chỉ là đặt lại PENDING kèm next_attempt_at.
• Transactional Outbox với FOR UPDATE SKIP LOCKED giữ DB và Kafka nhất quán mà không cần distributed transaction, đồng thời cho phép chạy nhiều relay song song.
• Độ ưu tiên được đảm bảo ở hai tầng: thứ tự poll trong SQL, và topic cùng concurrency riêng ở Kafka.
• Chống trùng nhiều lớp: Idempotency-Key ở API, kiểm tra trạng thái và khóa Redis ở worker, idempotency key gửi sang provider.
• Rate limiter dùng chung trên Redis bảo vệ quota của provider khi tải đột biến.

Từ khung này, bạn có thể mở rộng thêm template đa ngôn ngữ, webhook nhận trạng thái “đã giao/đã đọc” từ provider, hoặc dashboard theo dõi tỷ lệ FAILED theo từng kênh.