Spring Boot 实战 - 前端码易

创建时间: 2026-10-04最后更新: 2026-10-04作者: yangbo(8c2c979b6)

监控系统已经能够定时探测和保存结果，但用户还需要知道故障什么时候发生、什么时候恢复。如果每次 DOWN 都发一条通知，持续十分钟的故障就可能产生几十条重复消息；如果直接在事务里发送通知，数据库回滚后，用户仍可能收到一次根本没有保存的告警。

本章承接 RestClient、事务、Flyway 和定时任务章节，在 Java 21、Spring Boot 4.1.1、PostgreSQL 项目中实现告警闭环：连续三次失败时创建故障通知，恢复时创建恢复通知，通知任务与检查结果一起提交，再由后台投递。先理解应用事件的边界，再实现可以重试的 Outbox。

1. 应用事件不等于消息队列

普通对象就可以作为 Spring 应用事件，不需要继承 ApplicationEvent：

src/main/java/cn/usehook/spring/alert/HealthRecorded.java

1package cn.usehook.spring.alert;
2
3import cn.usehook.spring.health.HealthState;
4
5public record HealthRecorded(String service, HealthState state) {}

ApplicationEventPublisher.publishEvent() 把对象交给当前应用上下文的监听器。默认同步事件广播器在发布线程中调用 @EventListener：监听器慢，发布者也慢；监听器抛异常，可能使外层事务回滚。加 @Async 只是转交线程池，不会得到持久化、重启恢复或可靠投递能力。Spring 应用事件

只需要在提交之后记录辅助日志时，可以使用事务监听器：

src/main/java/cn/usehook/spring/alert/HealthRecordedLogger.java

01package cn.usehook.spring.alert;
02
03import org.slf4j.Logger;
04import org.slf4j.LoggerFactory;
05import org.springframework.stereotype.Component;
06import org.springframework.transaction.event.TransactionPhase;
07import org.springframework.transaction.event.TransactionalEventListener;
08
09@Component
10public class HealthRecordedLogger {
11    private static final Logger log = LoggerFactory.getLogger(HealthRecordedLogger.class);
12
13    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
14    public void onRecorded(HealthRecorded event) {
15        log.debug("Health recorded, service={}, state={}", event.service(), event.state());
16    }
17}

默认情况下，没有事务的发布不会触发这个监听器。事务回滚也不会触发 AFTER_COMMIT。但在数据库提交之后、监听器执行之前退出进程，事件就会丢失；它不能承担“必须最终发送”的通知职责。事务绑定事件

另外，AFTER_COMMIT 中看起来仍可访问原事务资源，不代表此时写入会再次提交。确实需要独立写库时，应通过另一个 Bean 开启 REQUIRES_NEW；这仍无法补上进程退出窗口。本章因此在原业务事务里直接写 Outbox，而不是提交之后再创建通知任务。TransactionalEventListener API

2. 定义告警状态和持久化任务

连续失败次数与当前故障编号属于独立状态。计数最高保存为 3，避免长时间故障不断累加；故障编号存在就表示已经触发过故障通知，后续 DOWN 不再创建新通知。恢复时沿用同一个故障编号，再清空当前故障。

检查序列	新建通知	保存后的状态
DOWN、DOWN	无	连续失败 2 次，没有故障编号
第三次 DOWN	FAILED	连续失败 3 次，产生故障编号
更多 DOWN	无	保持同一次故障
随后 UP	RECOVERED	失败次数清零，清除当前故障编号
更多 UP	无	保持正常

新增迁移 V7，不修改已执行的迁移。V6 留给资源级权限章节；Flyway 可以存在版本间隔，但已经部署 V7 后再补一个较低的 V6 需要重新规划迁移，不能随意开启乱序执行。

src/main/resources/db/migration/V7__create_alert_outbox.sql

01create table alert_state (
02    service varchar(50) primary key,
03    failures integer not null default 0 check (failures between 0 and 3),
04    incident_id uuid
05);
06
07create table alert_outbox (
08    id bigint generated always as identity unique,
09    event_id uuid primary key,
10    incident_id uuid not null,
11    service varchar(50) not null,
12    kind varchar(20) not null check (kind in ('FAILED', 'RECOVERED')),
13    created_at timestamptz not null default current_timestamp,
14    attempts integer not null default 0 check (attempts between 0 and 5),
15    available_at timestamptz not null default current_timestamp,
16    claim_token uuid,
17    locked_until timestamptz,
18    sent_at timestamptz,
19    dead_at timestamptz,
20    unique (incident_id, kind),
21    check ((claim_token is null) = (locked_until is null))
22);
23
24create index idx_alert_outbox_pending
25    on alert_outbox (available_at, id)
26    where sent_at is null and dead_at is null;
27create index idx_alert_outbox_service_order
28    on alert_outbox (service, id) where sent_at is null;

event_id 标识一条通知，重试保持不变；incident_id 把故障和恢复配成一组；id 用于同一服务的发送排序。unique (incident_id, kind) 防止同一故障重复产生同类型任务。dead_at 表示耗尽自动尝试，保留记录供人工处理，并不代表成功送达。

3. 检查结果和告警任务一起提交

替换 RestClient 章节的 HealthResultRecorder，保留 record(String, HealthState) 方法签名。HealthCheckService 仍然先在事务外调用目标，再调用这个 Bean。其他需要告警的检查入口也必须汇入它，不能继续只更新 health_status。

src/main/java/cn/usehook/spring/monitoring/HealthResultRecorder.java

01package cn.usehook.spring.monitoring;
02
03import cn.usehook.spring.alert.HealthRecorded;
04import cn.usehook.spring.health.HealthState;
05import cn.usehook.spring.health.HealthStatus;
06import java.util.Locale;
07import java.util.Objects;
08import java.util.UUID;
09import org.springframework.context.ApplicationEventPublisher;
10import org.springframework.jdbc.core.simple.JdbcClient;
11import org.springframework.stereotype.Service;
12import org.springframework.transaction.annotation.Transactional;
13
14@Service
15public class HealthResultRecorder {
16    private final JdbcClient jdbc;
17    private final ApplicationEventPublisher events;
18
19    public HealthResultRecorder(JdbcClient jdbc, ApplicationEventPublisher events) {
20        this.jdbc = jdbc;
21        this.events = events;
22    }
23
24    @Transactional
25    public HealthStatus record(String service, HealthState state) {
26        Objects.requireNonNull(state, "state");
27        String name = Objects.requireNonNull(service, "service").strip().toLowerCase(Locale.ROOT);
28        if (!name.matches("[a-z0-9][a-z0-9_-]{0,49}")) {
29            throw new IllegalArgumentException("Invalid service name");
30        }
31        jdbc.sql("""
32            insert into alert_state (service) values (:service)
33            on conflict (service) do nothing
34            """).param("service", name).update();
35        var previous = jdbc.sql("""
36            select failures, incident_id from alert_state
37            where service = :service for update
38            """).param("service", name)
39            .query((rs, row) -> new AlertState(
40                rs.getInt("failures"), rs.getObject("incident_id", UUID.class)
41            )).single();
42
43        int failures = state == HealthState.UP ? 0 : Math.min(3, previous.failures() + 1);
44        UUID incident = previous.incidentId();
45        String kind = null;
46        if (state == HealthState.DOWN && failures == 3 && incident == null) {
47            incident = UUID.randomUUID();
48            kind = "FAILED";
49        } else if (state == HealthState.UP && incident != null) {
50            kind = "RECOVERED";
51        }
52        jdbc.sql("""
53            insert into health_status (service, status) values (:service, :status)
54            on conflict (service) do update
55            set status = excluded.status, checked_at = current_timestamp
56            """).param("service", name).param("status", state.name()).update();
57        jdbc.sql("""
58            insert into health_check_history (service, status) values (:service, :status)
59            """).param("service", name).param("status", state.name()).update();
60        if (kind != null) {
61            jdbc.sql("""
62                insert into alert_outbox (event_id, incident_id, service, kind)
63                values (:event, :incident, :service, :kind)
64                """).param("event", UUID.randomUUID()).param("incident", incident)
65                .param("service", name).param("kind", kind).update();
66        }
67        jdbc.sql("""
68            update alert_state set failures = :failures, incident_id = :incident
69            where service = :service
70            """).param("failures", failures)
71            .param("incident", state == HealthState.UP ? null : incident, java.sql.Types.OTHER)
72            .param("service", name).update();
73        events.publishEvent(new HealthRecorded(name, state));
74        return new HealthStatus(state.name(), name);
75    }
76
77    private record AlertState(int failures, UUID incidentId) {}
78}

对 alert_state 的行锁使同一服务的计数计算串行执行，多个实例不会同时把“第二次失败”更新成两次独立的“第三次失败”。网络调用发生在加锁之前，锁只保护短暂的数据库操作。任何一条写入失败，最新状态、历史、失败计数与 Outbox 都一起回滚。

这里的连续顺序是“检查结果入库顺序”，不是远程探测开始时间。若允许同一目标同时执行多个探测，较早开始的任务可能较晚返回；应在后续多实例调度章节限制同一目标并行，或用采样序号拒绝旧结果。行锁本身不能判断网络结果的新旧。

4. 用短事务领取任务

领取任务后立即提交，随后在事务外发 HTTP。租约用于进程退出后的接管，随机 claim_token 区分两次领取，防止旧 worker 覆盖新 worker 的结果。

src/main/java/cn/usehook/spring/alert/AlertTask.java

1package cn.usehook.spring.alert;
2
3import java.time.OffsetDateTime;
4import java.util.UUID;
5
6public record AlertTask(
7    UUID eventId, UUID incidentId, String service, String kind,
8    OffsetDateTime occurredAt, int attempts, UUID claimToken
9) {}

src/main/java/cn/usehook/spring/alert/AlertOutbox.java

01package cn.usehook.spring.alert;
02
03import java.time.OffsetDateTime;
04import java.util.Optional;
05import java.util.UUID;
06import org.springframework.jdbc.core.simple.JdbcClient;
07import org.springframework.stereotype.Repository;
08import org.springframework.transaction.annotation.Transactional;
09
10@Repository
11public class AlertOutbox {
12    private final JdbcClient jdbc;
13
14    public AlertOutbox(JdbcClient jdbc) {
15        this.jdbc = jdbc;
16    }
17
18    @Transactional
19    public Optional<AlertTask> claim() {
20        jdbc.sql("""
21            update alert_outbox
22            set dead_at = clock_timestamp(), claim_token = null, locked_until = null
23            where sent_at is null and dead_at is null and attempts >= 5
24              and (locked_until is null or locked_until <= clock_timestamp())
25            """).update();
26        UUID token = UUID.randomUUID();
27        return jdbc.sql("""
28            with candidate as (
29                select job.event_id from alert_outbox job
30                where job.sent_at is null and job.dead_at is null and job.attempts < 5
31                  and job.available_at <= clock_timestamp()
32                  and (job.locked_until is null or job.locked_until <= clock_timestamp())
33                  and not exists (
34                      select 1 from alert_outbox older
35                      where older.service = job.service and older.id < job.id
36                        and older.sent_at is null
37                  )
38                order by job.id limit 1 for update of job skip locked
39            )
40            update alert_outbox job
41            set claim_token = :token, locked_until = clock_timestamp() + interval '30 seconds',
42                attempts = attempts + 1
43            from candidate where job.event_id = candidate.event_id
44            returning job.*
45            """).param("token", token)
46            .query((rs, row) -> new AlertTask(
47                rs.getObject("event_id", UUID.class), rs.getObject("incident_id", UUID.class),
48                rs.getString("service"), rs.getString("kind"),
49                rs.getObject("created_at", OffsetDateTime.class), rs.getInt("attempts"),
50                rs.getObject("claim_token", UUID.class)
51            )).optional();
52    }
53
54    @Transactional
55    public boolean finish(AlertTask task, boolean delivered) {
56        long delaySeconds = Math.min(300, 5L << (task.attempts() - 1));
57        return jdbc.sql("""
58            update alert_outbox
59            set sent_at = case when :delivered then clock_timestamp() else null end,
60                dead_at = case when not :delivered and attempts >= 5
61                    then clock_timestamp() else null end,
62                available_at = clock_timestamp() + (:delay * interval '1 second'),
63                claim_token = null, locked_until = null
64            where event_id = :event and claim_token = :token
65              and locked_until > clock_timestamp() and sent_at is null
66            """).param("delivered", delivered).param("delay", delaySeconds)
67            .param("event", task.eventId()).param("token", task.claimToken()).update() == 1;
68    }
69}

FOR UPDATE SKIP LOCKED 让不同 worker 跳过已被领取事务锁住的候选行。它适合队列领取，不适合用来读取完整业务视图。更新租约和增加尝试次数在同一个 SQL 中完成，任务不会在领取与更新之间被两个 worker 同时拿走。PostgreSQL SELECT 锁定语义

同一服务的旧任务未成功时，新任务不能发送。因此故障通知失败会阻挡恢复通知，而其他服务仍可继续。耗尽五次的任务也保留阻挡，运营人员需要检查接收方和失败日志，再明确重试或按业务流程处理，不能悄悄把恢复通知提前发送。

领取后即使尚未开始 HTTP 就退出，也会消耗一次尝试。这是有意选择的有界工作预算，避免反复崩溃的任务永远重启。第五次领取退出后，清理语句会在租约到期时将它标记为耗尽。

5. 在事务外发送并退避重试

沿用前文的 RestClient Starter 和 TaskConfiguration。通知地址来自运维配置，不接收用户传入的任意 URL；生产地址使用 HTTPS，凭据通过环境变量注入：

src/main/resources/application.yaml

1app:
2  alerts:
3    enabled: ${ALERTS_ENABLED:false}
4    webhook-url: ${ALERT_WEBHOOK_URL:http://localhost:8090/internal/alerts}
5    token: ${ALERT_WEBHOOK_TOKEN:}

默认关闭 worker，配置好接收端之后再开启。数据库仍然会保存待发送任务，避免“发送功能关闭”被误认为“通知不需要记录”。

src/main/java/cn/usehook/spring/alert/AlertWorker.java

01package cn.usehook.spring.alert;
02
03import java.net.URI;
04import java.net.http.HttpClient;
05import java.time.Duration;
06import java.time.OffsetDateTime;
07import java.util.UUID;
08import org.slf4j.Logger;
09import org.slf4j.LoggerFactory;
10import org.springframework.beans.factory.annotation.Value;
11import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
12import org.springframework.http.client.JdkClientHttpRequestFactory;
13import org.springframework.scheduling.annotation.Scheduled;
14import org.springframework.stereotype.Component;
15import org.springframework.web.client.RestClient;
16
17@Component
18@ConditionalOnProperty(name = "app.alerts.enabled", havingValue = "true")
19public class AlertWorker {
20    private static final Logger log = LoggerFactory.getLogger(AlertWorker.class);
21    private final AlertOutbox outbox;
22    private final RestClient client;
23    private final URI url;
24
25    public AlertWorker(
26        AlertOutbox outbox, RestClient.Builder builder,
27        @Value("${app.alerts.webhook-url}") URI url,
28        @Value("${app.alerts.token}") String token
29    ) {
30        if (token.isBlank()) throw new IllegalArgumentException("Alert token is required");
31        var http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2))
32            .followRedirects(HttpClient.Redirect.NEVER).build();
33        var factory = new JdkClientHttpRequestFactory(http);
34        factory.setReadTimeout(Duration.ofSeconds(3));
35        this.outbox = outbox;
36        this.url = url;
37        this.client = builder.requestFactory(factory)
38            .defaultHeader("Authorization", "Bearer " + token).build();
39    }
40
41    @Scheduled(fixedDelayString = "1s")
42    public void deliverOne() {
43        outbox.claim().ifPresent(task -> {
44            boolean delivered = false;
45            try {
46                var response = client.post().uri(url)
47                    .header("Idempotency-Key", task.eventId().toString())
48                    .body(new Notice(task.eventId(), task.incidentId(), task.service(),
49                        task.kind(), task.occurredAt()))
50                    .retrieve().toBodilessEntity();
51                delivered = response.getStatusCode().is2xxSuccessful();
52                if (!delivered) log.warn("Alert rejected, eventId={}, status={}",
53                    task.eventId(), response.getStatusCode());
54            } catch (RuntimeException failure) {
55                log.warn("Alert delivery failed, eventId={}", task.eventId(), failure);
56            }
57            if (!outbox.finish(task, delivered)) {
58                log.warn("Alert lease lost, eventId={}", task.eventId());
59            }
60        });
61    }
62
63    public record Notice(UUID eventId, UUID incidentId, String service,
64        String kind, OffsetDateTime occurredAt) {}
65}

连接和请求都有超时，30 秒租约留出正常调用与数据库确认的余量。但 GC 暂停、机器失联仍可能超过租约；原 worker 恢复后，带旧 token 的确认不会修改重新领取的任务。租约不可能撤回已经发往接收方的 HTTP，这也是仍需接收方去重的原因。

重试间隔为 5、10、20、40 秒，第五次失败后停止自动尝试。这里对所有失败统一保留有限重试，生产系统可将参数错误、认证失败直接转为人工处理，对 429/503 根据协议处理 Retry-After。不要在 HTTP 客户端与 Outbox 外层同时叠加无限重试。

如果 finish() 因数据库不可用而失败，任务保留原租约，之后可以重新领取。发送成功后、确认之前退出也会再次发送同一个 eventId。本例保证的是任务持久化和有界重试，不能宣称端到端恰好一次，也不保证在五次失败之后仍自动送达。

6. 接收方怎样去重

Idempotency-Key 只是协议字段，不会自动获得幂等效果。接收方必须校验身份、校验请求体和键是否一致，并为 eventId 建立唯一约束。若接收方把通知存成站内消息，可以在自己的数据库中使用以下表；这不是监控服务的 Flyway 迁移：

receiver/src/main/resources/db/migration/V1__create_notifications.sql

1create table received_notification (
2    event_id uuid primary key,
3    incident_id uuid not null,
4    service varchar(50) not null,
5    kind varchar(20) not null check (kind in ('FAILED', 'RECOVERED')),
6    occurred_at timestamptz not null
7);

经过接收方的身份校验、长度和枚举校验后，在 Service 事务中执行一次插入，提交成功或确认已存在之后返回 2xx：

receiver/src/main/java/cn/usehook/receiver/NotificationService.java

01@Transactional
02public void accept(Notice notice) {
03    int inserted = jdbc.sql("""
04        insert into received_notification (event_id, incident_id, service, kind, occurred_at)
05        values (:event, :incident, :service, :kind, :time)
06        on conflict (event_id) do nothing
07        """).param("event", notice.eventId()).param("incident", notice.incidentId())
08        .param("service", notice.service()).param("kind", notice.kind())
09        .param("time", notice.occurredAt()).update();
10    if (inserted == 0) {
11        long matching = jdbc.sql("""
12            select count(*) from received_notification
13            where event_id = :event and incident_id = :incident
14              and service = :service and kind = :kind and occurred_at = :time
15            """).param("event", notice.eventId()).param("incident", notice.incidentId())
16            .param("service", notice.service()).param("kind", notice.kind())
17            .param("time", notice.occurredAt()).query(Long.class).single();
18        if (matching != 1) throw new IllegalArgumentException("Event payload conflicts");
19    }
20}

这段是接收方 Service 方法片段：Notice 使用与发送方相同的 JSON 字段，JdbcClient 通过构造器注入，@Transactional 来自 org.springframework.transaction.annotation。重复请求只产生一条站内消息；接收方不能先返回 2xx 再尝试落库。相同 eventId 的不同内容会被拒绝，异常应由接收方的统一异常处理映射为 409。这里使用 PostgreSQL 默认的 READ COMMITTED 隔离级别，第二条查询能够看到与本次插入冲突的已提交记录。

如果接收端还要发送邮件或第三方即时消息，仅插入一个“已处理”标记再调用第三方仍可能丢消息；先调用再写标记则可能重复。接收方也需要持久化发送任务，或使用第三方提供的幂等键。去重必须覆盖实际副作用，不能只覆盖 HTTP Controller。

7. 验证阈值、回滚和接管

沿用 Testcontainers 的真实 PostgreSQL。下面的测试通过真实 Spring 事务代理调用 recorder；禁止为整个测试增加自动回滚事务，否则无法检查多次调用提交后的状态。

src/test/java/cn/usehook/spring/alert/AlertOutboxTests.java

001package cn.usehook.spring.alert;
002
003import cn.usehook.spring.TestcontainersConfiguration;
004import cn.usehook.spring.health.HealthState;
005import cn.usehook.spring.monitoring.HealthResultRecorder;
006import java.util.UUID;
007import org.junit.jupiter.api.Test;
008import org.springframework.beans.factory.annotation.Autowired;
009import org.springframework.boot.test.context.SpringBootTest;
010import org.springframework.context.annotation.Import;
011import org.springframework.jdbc.core.simple.JdbcClient;
012import org.springframework.security.oauth2.jwt.JwtDecoder;
013import org.springframework.test.annotation.DirtiesContext;
014import org.springframework.test.context.bean.override.mockito.MockitoBean;
015import org.springframework.transaction.PlatformTransactionManager;
016import org.springframework.transaction.support.TransactionTemplate;
017import static org.assertj.core.api.Assertions.assertThat;
018
019@SpringBootTest(properties = {
020    "app.alerts.enabled=false",
021    "app.monitoring.initial-delay=1h",
022    "app.monitoring.jobs.enabled=false",
023    "app.history.retention.enabled=false"
024})
025@Import(TestcontainersConfiguration.class)
026@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
027class AlertOutboxTests {
028    @Autowired HealthResultRecorder recorder;
029    @Autowired AlertOutbox outbox;
030    @Autowired JdbcClient jdbc;
031    @Autowired PlatformTransactionManager transactions;
032    @MockitoBean JwtDecoder jwtDecoder;
033
034    @Test
035    void oneIncidentCreatesOnlyFailureAndRecovery() {
036        String service = "test-" + UUID.randomUUID();
037        recorder.record(service, HealthState.DOWN);
038        recorder.record(service, HealthState.DOWN);
039        assertThat(count(service, "alert_outbox")).isZero();
040        recorder.record(service, HealthState.DOWN);
041        recorder.record(service, HealthState.DOWN);
042        recorder.record(service, HealthState.UP);
043        recorder.record(service, HealthState.UP);
044        assertThat(count(service, "alert_outbox")).isEqualTo(2);
045        assertThat(count(service, "health_check_history")).isEqualTo(6);
046        assertThat(jdbc.sql("select kind from alert_outbox where service = :service order by id")
047            .param("service", service).query(String.class).list())
048            .containsExactly("FAILED", "RECOVERED");
049
050        var tx = new TransactionTemplate(transactions);
051        tx.executeWithoutResult(status -> {
052            for (int i = 0; i < 3; i++) recorder.record(service, HealthState.DOWN);
053            status.setRollbackOnly();
054        });
055        assertThat(count(service, "alert_outbox")).isEqualTo(2);
056        assertThat(count(service, "health_check_history")).isEqualTo(6);
057        assertThat(jdbc.sql("select failures from alert_state where service = :service")
058            .param("service", service).query(Integer.class).single()).isZero();
059    }
060
061    @Test
062    void expiredLeaseCannotAcknowledgeAndRetriesAreBounded() {
063        String service = "lease-" + UUID.randomUUID();
064        for (int i = 0; i < 3; i++) recorder.record(service, HealthState.DOWN);
065        recorder.record(service, HealthState.UP);
066        var first = outbox.claim().orElseThrow();
067        assertThat(first.kind()).isEqualTo("FAILED");
068        assertThat(outbox.claim()).isEmpty();
069        jdbc.sql("""
070            update alert_outbox set locked_until = clock_timestamp() - interval '1 second'
071            where event_id = :event
072            """).param("event", first.eventId()).update();
073        var replacement = outbox.claim().orElseThrow();
074        assertThat(replacement.claimToken()).isNotEqualTo(first.claimToken());
075        assertThat(outbox.finish(first, true)).isFalse();
076        assertThat(outbox.finish(replacement, true)).isTrue();
077        for (int attempt = 1; attempt <= 5; attempt++) {
078            var recovery = outbox.claim().orElseThrow();
079            assertThat(recovery.kind()).isEqualTo("RECOVERED");
080            assertThat(recovery.incidentId()).isEqualTo(first.incidentId());
081            assertThat(recovery.attempts()).isEqualTo(attempt);
082            assertThat(outbox.finish(recovery, false)).isTrue();
083            jdbc.sql("""
084                update alert_outbox set available_at = clock_timestamp() - interval '1 second'
085                where event_id = :event
086                """).param("event", recovery.eventId()).update();
087        }
088        assertThat(outbox.claim()).isEmpty();
089        assertThat(jdbc.sql("""
090            select count(*) from alert_outbox
091            where service = :service and dead_at is not null and attempts = 5
092            """).param("service", service).query(Long.class).single()).isEqualTo(1);
093    }
094
095    private long count(String service, String table) {
096        // table 仅来自上面两个测试常量，不接收外部输入。
097        return jdbc.sql("select count(*) from " + table + " where service = :service")
098            .param("service", service).query(Long.class).single();
099    }
100}

@DirtiesContext 让每个方法重新创建上下文和其 PostgreSQL 容器，确保领取测试面对空队列，不误取另一条测试留下的任务。它适合这里少量故障场景，不应成为全项目每条测试的默认配置。本机具备容器运行时后可执行 ./mvnw -Dtest=AlertOutboxTests test。这些测试验证状态机、事务、租约和重试上限，不证明网络投递可靠；还应按以下故障序列在独立测试数据库和可控接收端演练：

场景	必须观察到的结果
接收方连续返回 503	attempts 最终为 5，dead_at 非空，停止自动领取
发送成功后模拟确认失败	接收方可能收到重复 HTTP，但同一 eventId 只有一条消息
worker 领取后退出	30 秒后另一 worker 能重新领取，claim_token 改变
旧 worker 再调用 finish	返回 false，不能覆盖新 worker 的结果
故障通知仍失败、服务已经恢复	恢复任务保留待发，不越过前面的故障任务

运行时监控待发送数量、最老任务年龄、耗尽数量和发送失败率。恢复故障后由受控运维操作重置指定任务的尝试次数及可用时间，并保留原 eventId；不要删除失败记录、重新生成 ID 后假装它是新通知。已成功通知的清理周期也必须与接收方幂等记录保留期协调，人工重放不能超出去重窗口。

首页订阅