Entwurf eines Benachrichtigungssystems: Push, Email, SMS mit Retry
1. Überblick
Praktisch jedes Produkt braucht Benachrichtigungen: OTP-Codes per SMS, Bestätigungsmails für Bestellungen, Push-Benachrichtigungen für neue Nachrichten. Eine Benachrichtigung zu versenden ist einfach. Schwierig wird es, Millionen zu versenden, ohne sie zu verlieren, ohne Duplikate, und damit OTP-Codes nicht hinter Marketing-E-Mails anstehen.
In diesem Artikel entwerfen wir ein Multi-Channel-Benachrichtigungssystem von Grund auf. Wir behandeln Anforderungen, Datenmodell, API, Verarbeitungsflüsse und Lösungen für knifflige Probleme: doppelte Sendungen, Konsistenz zwischen DB und Kafka, Burst-Lasten. Der Code verwendet Java 21, Spring Boot 3.x, PostgreSQL, Kafka und Redis.
2. Anforderungen
2.1. Funktionale Anforderungen
- Andere Services (Order, Auth, Marketing…) rufen eine API auf, um Benachrichtigungen an einen Benutzer über einen oder mehrere Kanäle zu versenden: PUSH, EMAIL, SMS.
- Jede Benachrichtigung hat eine Priorität:
HIGH(OTP, Sicherheitswarnungen),NORMAL(Transaktionen),LOW(Marketing). - Das System wiederholt den Versand automatisch, wenn ein Provider einen vorübergehenden Fehler verursacht. Nach maximalen Versuchen wird die Benachrichtigung als
FAILEDmarkiert. - Das System respektiert Benutzereinstellungen, zum Beispiel, wenn ein Benutzer Marketing-E-Mails deaktiviert hat.
2.2. Skalierung und Einschränkungen
- Etwa 50 Millionen Benachrichtigungen pro Tag. Im Durchschnitt etwa 600 pro Sekunde, bei Spitzenlast während Marketing-Kampagnen bis zu 20.000 pro Sekunde.
- Benachrichtigungen mit
HIGH-Priorität müssen innerhalb weniger Sekunden versendet werden, auch bei Spitzenlast. - Provider für SMS/E-Mail haben Ratenlimits, zum Beispiel 500 Anfragen pro Sekunde pro API-Schlüssel.
- Die Semantik auf der Infrastruktur-Ebene ist at-least-once, aber wir müssen Duplikate in angemessenem Umfang verhindern. Ein Benutzer erhält zwei OTP-Codes = schlechte Erfahrung. Zwei SMS erhalten bedeutet auch doppelte Kosten.
3. Gesamtarchitektur
Client-Service ──POST──▶ Notification API ──(1 Transaktion)──▶ PostgreSQL
│ notification, delivery (PENDING)
▼
Outbox Relay (Poll, SKIP LOCKED)
│ Veröffentliche nach Priorität
┌───────────────────────────────┼───────────────────────────┐
▼ ▼ ▼
notifications.high notifications.normal notifications.low
│ │ │
└────────────▶ Delivery Workers ◀───────────────────────────┘
│ Dedup (Redis) + Rate Limit (Redis)
▼
FCM/APNs · Email-Provider · SMS-Provider
│
Erfolgreich → SENT │ Fehler → PENDING + next_attempt_at
Die Kernidee: Die delivery-Tabelle ist die Quelle der Wahrheit. Kafka ist nur das Transportmedium. Retry benötigt kein separates Retry-Topic. Der Worker muss nur den Datensatz auf PENDING mit next_attempt_at in der Zukunft zurücksetzen, und das Outbox Relay wird ihn später erneut pushen, wenn die Zeit abgelaufen ist.
4. Datenmodell
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';
Einige wichtige Punkte:
idempotency_key UNIQUEverhindert, dass der Client die Benachrichtigungs-API erneut aufruft und doppelte Benachrichtigungen erstellt.- Jeder Kanal ist eine eigene
delivery-Zeile. Wenn E-Mail fehlschlägt, muss Push nicht erneut gesendet werden. prioritywird indeliverywiederholt, um den partiellen Indexidx_delivery_duegenau für die Abfrage des Relay zu nutzen. Der Index enthält nur Zeilen mitPENDING, daher bleibt er klein, auch wenn die Tabelle Milliarden von Zeilen mitSENTenthält.
5. API zum Empfangen von Benachrichtigungen
Zuerst deklarieren Sie die benötigten Abhängigkeiten:
<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>
Dann die API. Der Client muss den Header Idempotency-Key mitsendet. Normalerweise ist dies eine Geschäfts-ID, zum Beispiel 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));
}
}
Die API gibt 202 Accepted zurück, da das Versenden asynchron erfolgt. Die eigentliche Verarbeitung findet im Service statt:
@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()) { // wiederholte Anfrage → gebe alte Benachrichtigung zurück
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 ungültig", e);
}
}
}
ON CONFLICT ... DO NOTHING RETURNING id bearbeitet Konflikte direkt in der Datenbank. Wenn zwei Anfragen mit demselben Schlüssel gleichzeitig ankommen, schlägt nur eine Insert fehl. Die andere Anfrage erhält eine leere Liste zurück und liest die alte ID erneut. Sie benötigen keine Lock auf der Anwendungsebene.
Beachten Sie, dass der Service nicht Kafka innerhalb dieser Transaktion sendet. Das ist das Transactional-Outbox-Muster, das im nächsten Abschnitt erläutert wird.
6. Outbox Relay und Prioritätswarteschlange
Angenommen, Sie schreiben die DB und senden dann sofort Kafka. Wenn die Anwendung dazwischen abstürzt, sitzt die Benachrichtigung in der DB, wird aber nie gesendet. Wenn Sie es umgekehrt machen – Kafka zuerst, dann DB – kann ein Worker einen noch nicht existierenden Datensatz verarbeiten. Outbox löst dieses Problem: Die DB ist der einzige Ort, an den in der Transaktion geschrieben wird, und das Relay liest die DB und veröffentlicht danach.
@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);
}
}
}
Der obige Code macht drei Dinge:
ORDER BY priority, next_attempt_atwählt die Zeilen mitHIGH-Priorität zuerst aus. Auch wenn 5 Millionen Marketing-E-Mails warten, wird eine neue OTP beim nächsten Poll-Zyklus ausgewählt.FOR UPDATE SKIP LOCKEDermöglicht das parallele Ausführen mehrerer Relay-Instanzen. Jede Instanz sperrt einen anderen Satz und ignoriert die Zeilen, die andere Instanzen halten. Keine Instanz muss auf eine andere warten..join()wartet auf die Broker-Bestätigung, bevor der Status geändert wird. Wenn das Relay nach dem Senden, aber vor dem Commit abstürzt, bleibt die Zeile aufPENDINGund wird erneut gesendet. Das ist at-least-once. Der Worker muss gegen Duplikate schützen, siehe Abschnitt 7.2.
Die Priorität wird noch einmal auf der Consumer-Seite durch separate Topics und getrennte Concurrency durchgesetzt:
@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 hat keine Priority Queue innerhalb eines Topics. Deshalb ist die Trennung der Topics der praktischste Ansatz. Wenn die Kampagne notifications.low mit Millionen von Nachrichten überlastet wird, läuft der Consumer von notifications.high immer noch auf separaten Threads und wird nicht beeinträchtigt. Die Anzahl der concurrency muss kleiner als oder gleich der Anzahl der Partitionen des entsprechenden Topics sein.
7. Worker zum Versenden und Retry
7.1. Retry-Richtlinie
Nicht jeder Fehler sollte wiederholt werden. Timeouts oder HTTP 429/503 sind vorübergehende Fehler. Eine falsche Telefonnummer oder ein widerrufener Push-Token sind permanente Fehler; Retry verschwendet nur Geld.
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: Anzahl der bisherigen Versuche (beginnend bei 1). Leer = Retry beenden. */
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())));
}
}
Mit base = 2s, max = 5m, maxAttempts = 6 sind die Wartezeiten nacheinander:
1. Versuch Fehler → Warte 2s
2. Versuch Fehler → Warte 4s
3. Versuch Fehler → Warte 8s
4. Versuch Fehler → Warte 16s
5. Versuch Fehler → Warte 32s
6. Versuch Fehler → FAILED
In der Produktion sollten Sie einen zufälligen Jitter zu den Wartezeiten hinzufügen. Ansonsten werden Tausende von Benachrichtigungen mit demselben Fehler genau zum gleichen Zeitpunkt erneut versucht, wenn der Provider ausfällt. Dieses Phänomen wird „Thundering Herd” genannt.
Sie können diese Richtlinie mit einem einfachen Test überprüfen:
class RetryPolicyTest {
private final RetryPolicy policy =
new RetryPolicy(Duration.ofSeconds(2), Duration.ofSeconds(10), 6);
@Test
void backoffWaechstExponentielUndWirdBegrenzt() {
assertThat(policy.nextDelay(1)).contains(Duration.ofSeconds(2));
assertThat(policy.nextDelay(3)).contains(Duration.ofSeconds(8));
assertThat(policy.nextDelay(4)).contains(Duration.ofSeconds(10)); // wird durch max begrenzt
}
@Test
void stoppRetryBeiErreichenDerMaximalenVersuche() {
assertThat(policy.nextDelay(6)).isEmpty();
}
}
7.2. Worker: Dedup, Ratenlimit, Statusupdate
public class TransientSendException extends RuntimeException {
public TransientSendException(String msg) { super(msg); }
}
public interface ChannelSender {
Channel channel();
void send(UUID deliveryId); // wirft TransientSendException bei vorübergehendem Fehler
}
@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; // doppelte Message von Kafka
Boolean first = redis.opsForValue()
.setIfAbsent("notif:sending:" + id, "1", Duration.ofMinutes(10));
if (!Boolean.TRUE.equals(first)) return; // anderer Worker versendet
Channel channel = Channel.valueOf((String) d.get("channel"));
int attempt = ((Number) d.get("attempts")).intValue() + 1;
try {
if (!acquireRate(channel)) {
throw new TransientSendException("ratenlimitiert"); // zurückschieben, nicht als Provider-Fehler zählen
}
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) { // permanenter Fehler
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;
}
}
Der Worker hat zwei Schutzschichten gegen Duplikate:
- Überprüfung von
status = 'SENT'eliminiert Messages, die nach erfolgreichem Versand erneut veröffentlicht wurden. - Redis-Lock
SET NXmit TTL verhindert, dass zwei Consumer die gleiche Message empfangen, zum Beispiel bei Kafka-Rebalance in der Mitte. Der TTL von 10 Minuten stellt sicher, dass der Lock sich selbst freigeben, wenn der Worker abstürzt.
Das ist immer noch nicht exactly-once. Wenn der Worker den Provider erfolgreich aufgerufen hat, aber vor dem UPDATE ... 'SENT' abstürzt, kann die Benachrichtigung erneut versendet werden, nachdem der Lock abgelaufen ist. Um das zu vermeiden, übergeben Sie deliveryId als Idempotency-Schlüssel an den Provider, wenn dieser es unterstützt. Viele E-Mail-Provider und FCM haben einen solchen Mechanismus.
acquireRate ist ein Fixed-Window-Rate-Limiter auf Redis. Alle Worker-Instanzen teilen sich einen Zähler pro Sekunde für jeden Kanal, sodass die Gesamtgeschwindigkeit das 500-request/s-Quota des Providers nicht überschreitet. Wenn das Limit überschritten wird, wird die Benachrichtigung auf PENDING zurückgesetzt, anstatt den Provider aufzurufen und HTTP 429 zu erhalten.
8. Schwierige Punkte und Lösungen
8.1. Burst-Lasten
Wenn Marketing 10 Millionen Benachrichtigungen auf einmal versendet, trägt das System die Last auf mehreren Ebenen:
- API nur insert in DB und return, daher kann es hohe Lasten tragen. Bei großen Kampagnen sollten Sie eine separate Batch-API bereitstellen, die eine Liste von Benutzern oder ein Segment empfängt. Ein Background-Job generiert dann die
delivery-Zeilen in Losen von 1.000. - Kafka nimmt die Reste auf. Topic
lowmit einem großen Rückstand ist normal, solange Topichighnoch fließt. - Ratenlimiter schützt den Provider. Benachrichtigungen mit
LOW-Priorität etwas langsamer zu versenden ist nicht schlecht.
Sie sollten eine zusätzliche Maßnahme hinzufügen: Reservieren Sie in acquireRate einen Teil des Quotes speziell für HIGH, zum Beispiel darf LOW nur 60% des Limits nutzen. So wird OTP nie durch Marketing-E-Mails ratenlimitiert.
8.2. Konsistenz zwischen DB und Kafka
Outbox kombiniert mit at-least-once und idempotenten Consumern ist der Standard-Triple. Sie benötigen keine Distributed Transaction (XA). Der Preis ist zusätzliche Latenz von etwa einer Poll-Periode, hier 200ms. Wenn Sie eine noch niedrigere Latenz benötigen, ersetzen Sie Polling durch CDC (WAL-Lesezugriff auf PostgreSQL), aber die Infrastruktur wird komplexer.
8.3. Datenwachstum
Die delivery-Tabelle wächst pro Tag um etwa 60–80 Millionen Zeilen. Sie sollten die Tabelle nach Tag partitionieren (PARTITION BY RANGE (created_at)) und ältere Partitionen nach 30–90 Tagen löschen. Der partielle Index aus Abschnitt 4 hält die Abfrage des Relay schnell, ersetzt aber nicht die Notwendigkeit, alte Daten zu bereinigen.
8.4. Benutzereinstellungen und Ruhezeiten
Überprüfen Sie die Einstellungen direkt beim Erstellen von delivery in NotificationService. Kanäle, die der Benutzer deaktiviert hat, erhalten keine Zeile. Bei Ruhezeiten, zum Beispiel keine Marketing-Push nach 22 Uhr, setzen Sie einfach next_attempt_at auf 8 Uhr morgens des nächsten Tages. Der Relay-Mechanismus kümmert sich um den Rest, kein zusätzlicher Scheduler nötig.
Anmerkung: Rufen Sie keine Provider-APIs innerhalb langer
@Transactional-Methoden auf. Ein langsamer HTTP-Request von 30 Sekunden hält die DB-Verbindung und Row-Locks für die ganze Zeit, und der Pool erschöpft sich sehr schnell. InDeliveryWorkerist jedesUPDATEeine kurze, unabhängige Transaktion, getrennt vom Provider-Aufruf.
Anmerkung: Versuchen Sie nicht unendlich zu wiederholen und wiederholen Sie keine permanenten Fehler. Eine falsche Telefonnummer, 6 Retry-Versuche, multipliziert mit 1 Million Benutzern = riesige SMS-Rechnung. Klassifizieren Sie die Fehlercodes jedes Providers sorgfältig.
8.5. Wann Sie diesen Entwurf NICHT verwenden sollten
Wenn Sie nur ein paar Tausend E-Mails pro Tag versenden, sind PostgreSQL plus SKIP LOCKED ausreichend. Der Worker pollt die delivery-Tabelle direkt, Kafka entfällt ganz. Kafka lohnt sich nur, wenn Sie Consumer nach Priorität auf großem Maßstab trennen müssen, oder wenn mehrere andere Systeme den gleichen Event-Stream konsumieren. Umgekehrt: Wenn Sie absolute Reihenfolge zwischen Benachrichtigungen desselben Benutzers gewährleisten müssen, reicht dieser Entwurf nicht. Sie müssen userId als Kafka-Key verwenden und asynchrone Retry für diese Benachrichtigungen deaktivieren.
9. Fazit
Wir haben ein Multi-Channel-Benachrichtigungssystem mit folgenden Kernentscheidungen entworfen:
- Die
delivery-Tabelle ist die Quelle der Wahrheit. Retry bedeutet nur,PENDINGmitnext_attempt_atin der Zukunft zurückzusetzen. - Transactional Outbox mit
FOR UPDATE SKIP LOCKEDhält DB und Kafka konsistent, ohne Distributed Transaction, und erlaubt mehrere parallele Relay-Instanzen. - Priorität wird auf zwei Ebenen gewährleistet: Poll-Reihenfolge in SQL, und getrennte Topics mit getrenner Concurrency in Kafka.
- Multi-Layer-Dedup:
Idempotency-Keyin der API, Statusprüfung und Redis-Lock im Worker, Idempotency-Key an Provider. - Gemeinsamer Ratenlimiter auf Redis schützt das Provider-Quota bei Burst-Lasten.
Aus diesem Framework können Sie weitere Funktionen hinzufügen: Multi-Sprachen-Templates, Webhooks für „Zugestellt/Gelesen”-Status vom Provider, oder ein Dashboard zur Überwachung der FAILED-Quote pro Kanal.