通知システム設計: push、email、SMSを備えたretryと優先度

1. 概要

ほぼすべての製品は通知を送信する必要があります。SMS経由のOTPコード、注文確認メール、新しいメッセージが到着したときのプッシュ通知など。1つの通知を送るのは簡単です。難しいのは、数百万の通知を送信する際に、失われず、重複せず、OTPがマーケティングメールの後に待たされないようにすることです。

この記事では、エンドツーエンドでマルチチャネル通知システムを設計します。対象範囲は、要件、データモデル、API、処理フロー、および難しい点を解決する方法です: 重複送信、DBとKafkaの一貫性、突発的な負荷。コード例にはJava 21、Spring Boot 3.x、PostgreSQL、Kafka、Redisを使用します。

2. 問題の要件

2.1. 機能要件

• 他のサービス (Order、Auth、Marketing など) がAPIを呼び出して、1つ以上のチャネル (PUSH、EMAIL、SMS) を通じて1ユーザーに通知を送信します。
• 各通知には優先度があります: HIGH (OTP、セキュリティアラート)、NORMAL (トランザクション)、LOW (マーケティング)。
• システムはプロバイダーの一時的なエラーを自動的に再試行します。最大回数に達したら、通知は FAILED とマークされて追跡されます。
• システムはユーザーの設定を尊重します。例えば、ユーザーがマーケティングメールをオフにしている場合など。

2.2. スケールと制約

• 1日あたり約5000万の通知。平均約600/秒、マーケティングキャンペーン実行時のピークは20,000/秒に達することがあります。
• HIGH優先度の通知は、ピーク負荷時であっても数秒以内に送信される必要があります。
• SMS/メールプロバイダーはレート制限を設定します。例えば、APIキーあたり1秒あたり500リクエスト。
• 配信セマンティクスはインフラストラクチャレベルではat-least-onceですが、合理的なレベルで重複を防ぐ必要があります。ユーザーが同じOTPを2回受け取るのは悪い体験です。2回のSMSを受け取ると課金されるため、コストもかかります。

3. 全体アーキテクチャ

クライアントサービス ──POST──▶ 通知API ──(1トランザクション)──▶ PostgreSQL
                                                         │ notification, delivery (PENDING)
                                                         ▼
                                              Outbox Relay (poll, SKIP LOCKED)
                                                         │ 優先度に基づいて発行
                         ┌───────────────────────────────┼───────────────────────────┐
                         ▼                               ▼                           ▼
              notifications.high              notifications.normal          notifications.low
                         │                               │                           │
                         └────────────▶ 配信ワーカー ◀───────────────────────────┘
                                         │  重複排除 (Redis) + レート制限 (Redis)
                                         ▼
                              FCM/APNs · メールプロバイダー · SMSプロバイダー
                                         │
                       成功 → SENT │ 一時的なエラー → PENDING + next_attempt_at

主なアイデア: delivery テーブルが真実の源です。Kafkaは単なる転送メカニズムです。retryは専用のretry topicを必要としません。ワーカーは単にレコードを 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 行です。メールエラーがプッシュの再送信を引き起こしません。
• priority は delivery で重複しており、Relayのクエリにサービスを提供するための部分インデックス idx_delivery_due をサポートします。インデックスは 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
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("ペイロードが無効です", e);
        }
    }
}

ON CONFLICT ... DO NOTHING RETURNING id はデータベース内で同時性を直接処理します。同じキーで2つのリクエストが同時に到着した場合、1つのリクエストのみがinsertに成功します。もう一方のリクエストは空のリストを受け取り、古いIDを読み返します。アプリケーション層でロックする必要はありません。

このサービスは、このトランザクション内でKafkaを送信しないことに注意してください。これがTransactional Outboxパターンです。次のセクションで理由を説明します。

6. Outbox Relayと優先度付きキュー

DBに書き込んですぐ後にKafkaを送信するとします。アプリケーションがこの2つのステップの間に死亡した場合、通知はDBに存在しますが、送信されることはありません。逆にしても同じ問題が発生します。Outboxはこれを解決します: DBが唯一の書き込み先で、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);
        }
    }
}

上記のコードは3つのことを行います:

• ORDER BY priority, next_attempt_at は HIGH 行を最初に取得します。待機中の500万のマーケティングメールがある場合でも、新しいOTPは次のポーリングサイクルで選択されます。
• FOR UPDATE SKIP LOCKED により、複数のRelayインスタンスが並列で実行できます。各インスタンスは異なるバッチをロックし、他のインスタンスが保持している行をスキップします。どのインスタンスも互いに待つ必要はありません。
• .join() はブローカー確認を待ってからステータスを変更します。Relayが送信後に死亡しがコミット前に、行は PENDING のままで再送信されます。これはat-least-onceです。ワーカーは重複を防ぐ必要があります。セクション7.2を参照してください。

優先度は、専用トピックと専用コンカレンシーを使用してコンシューマー側で再度保証されます:

@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は単一トピック内で優先度キューを持っていません。したがって、トピックの分割が最も現実的な方法です。マーケティング キャンペーンが notifications.low に数百万のメッセージを蓄積した場合、notifications.high のコンシューマーは別のスレッドで実行され、影響を受けません。concurrency 数は、対応するトピックのパーティション数以下である必要があります。

7. ワーカーの送信とretry

7.1. Retry ポリシー

すべてのエラーを再試行すべきではありません。タイムアウトまたはHTTP 429/503は一時的なエラーです。間違った電話番号またはプッシュトークンの取り消しは永続的なエラーで、retryはお金を浪費するだけです。

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 = 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())));
    }
}

base = 2s、max = 5m、maxAttempts = 6 では、待機間隔は次の通りです:

1回目エラー → 2s待機
2回目エラー → 4s待機
3回目エラー → 8s待機
4回目エラー → 16s待機
5回目エラー → 32s待機
6回目エラー → FAILED

本番環境では、待機間隔にランダムなジッターを追加する必要があります。そうしないと、プロバイダーが故障している同じエラーを持つ数千の通知が、同じ時刻に再試行します。この現象は thundering herd と呼ばれます。

このポリシーは簡単なテストで検証できます:

class RetryPolicyTest {

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

    @Test
    void backoffIsExponentialAndCappedByMax() {
        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 stopRetryWhenMaxAttemptsReached() {
        assertThat(policy.nextDelay(6)).isEmpty();
    }
}

7.2. ワーカー: 重複排除、レート制限、ステータス更新

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からの重複メッセージ

        Boolean first = redis.opsForValue()
            .setIfAbsent("notif:sending:" + id, "1", Duration.ofMinutes(10));
        if (!Boolean.TRUE.equals(first)) return;                  // 別のワーカーが送信中

        Channel channel = Channel.valueOf((String) d.get("channel"));
        int attempt = ((Number) d.get("attempts")).intValue() + 1;
        try {
            if (!acquireRate(channel)) {
                throw new TransientSendException("レート制限中");  // 延期、プロバイダーエラーとしてカウントしない
            }
            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;
    }
}

ワーカーには2つのレイヤーの重複排除があります:

1. status = 'SENT' をチェックして、送信完了後に再度発行されたメッセージを削除します。
2. Redis SET NX with TTL は、例えばKafkaの rebalance 中に2つのコンシューマーが同じメッセージを受け取る場合をブロックします。TTL 10分は、ワーカーが死亡した場合にロックが自動的に解放されるようにします。

これはまだ exactly-once ではありません。ワーカーがプロバイダーを正常に呼び出してから UPDATE ... 'SENT' する前に死亡した場合、ロックの有効期限が切れた後に通知が再送信される可能性があります。より厳密にするには、deliveryId をべき等キーとしてプロバイダーに渡します。プロバイダーがこれをサポートしている場合です。多くのメール プロバイダーとFCMはこのメカニズムを持っています。

acquireRate は Redis上の fixed window タイプのレート制限です。すべてのワーカー インスタンスはチャネルごとに共有カウンターを使用するため、総スループットはプロバイダーのクォータ500 request/秒を超えません。閾値を超えた場合、通知はプロバイダーを呼び出して HTTP 429 を受け取る代わりに、PENDING に戻されます。

8. 難しい点と解決策

8.1. 突発的な負荷

マーケティングが1000万の通知を一度に送信する場合、システムは各レイヤーで負荷を処理します:

• API はDBに insert して返すだけなので、高負荷に対応できます。大きなキャンペーンの場合は、ユーザーのリストまたはセグメントを受け取る別のバッチAPI を提供する必要があります。バックグラウンド ジョブは1,000行のロットで delivery 行を生成します。
• Kafka はオーバーフロー分を吸収します。low トピックが処理待ちになるのは普通のことで、high トピックが流れ続ける限り問題ありません。
• レート制限がプロバイダーを保護します。LOW 通知をもう少し遅く送信しても誰も文句を言いません。

さらに1つの対策を追加する必要があります: acquireRate で、HIGH 用に引き当てられたクォータを確保します。例えば、LOW はしきい値の60%のみを使用できます。このようにして、OTPはメール マーケティングのレート制限によって決してブロックされません。

8.2. DBとKafkaの一貫性

Outbox と at-least-once、そしてべき等コンシューマーを組み合わせるのが標準的な3点セットです。分散トランザクション (XA) は不要です。支払う代償はポーリング周期の1サイクル分の遅延で、ここでは200msです。より低い遅延が必要な場合は、ポーリングをCDC (PostgreSQLのWALを読む) に置き換えますが、インフラストラクチャはより複雑になります。

8.3. データの肥大化

毎日、delivery テーブルに約6000〜8000万行が追加されます。テーブルを日ごとに partition し (PARTITION BY RANGE (created_at))、30〜90日以上前のパーティションをドロップする必要があります。セクション4の部分インデックスはリレーのクエリを常に高速に保ちますが、データ クリーンアップの代わりにはなりません。

8.4. ユーザーの設定と静かな時間

NotificationService で delivery を作成するときに、ユーザーの設定を直接確認します。ユーザーがオフにしたチャネルの行は作成しないでください。クワイエット アワー (例えば22時以降にマーケティング プッシュを送信しない) の場合、next_attempt_at を次の朝8時に設定するだけです。relay メカニズムが残りを処理し、他のスケジューラは必要ありません。

注意: トランザクション内でプロバイダーを呼び出さないでください。遅いHTTPリクエスト (30秒) はDBコネクションと行ロックを保持し、プール がすぐに枯渇します。DeliveryWorker では、各 UPDATE は短くて独立したトランザクションで、プロバイダーの呼び出しとは無関係です。

注意: 無制限に再試行したり、永続的なエラーを再試行したりしないでください。間違った電話番号を6回再試行 × 100万ユーザーは、非常に大きいSMS請求書になります。各プロバイダーのエラーコードを慎重に分類してください。

8.5. このデザインを使用してはいけない場合

1日に数千通のメールのみを送信する場合、PostgreSQL + SKIP LOCKED で十分です。ワーカーは delivery テーブルを直接ポーリングし、Kafkaは完全に削除します。Kafkaは、優先度別にコンシューマーを分割する価値がある大規模時か、複数のシステムが同じイベント ストリームを消費する必要がある場合のみです。逆に、同じユーザーの通知間で絶対的な順序を保証する必要がある場合、上記のデザインは不十分です。Kafkaキーとして userId を使用し、非同期retryを削除する必要があります。

9. まとめ

エンドツーエンドのマルチチャネル通知システムを設計しました。主な決定は以下の通りです:

• delivery テーブルが真実の源。 Retry は PENDING に設定してから next_attempt_at を設定することだけです。
• Transactional Outbox + FOR UPDATE SKIP LOCKED はDBとKafkaの一貫性を保ち、分散トランザクションを必要とせず、複数のRelayインスタンスを並列実行できます。
• 優先度は2つのレイヤーで保証されます: SQLでのポーリング順序と、Kafkaでの専用トピック + コンカレンシー。
• 多層重複排除: APIでの Idempotency-Key、ワーカーでのステータスチェックとRedisロック、プロバイダーへのべき等キーの送信。
• Redisでの共有レート制限 は、突発的な負荷時にプロバイダーのクォータを保護します。

このフレームワークから、多言語テンプレート、プロバイダーからの「配信/読了」ステータスの webhook、またはチャネルあたりの FAILED 比率を追跡するダッシュボードまで、さらに拡張できます。