Spring Boot 实战 - 前端码易

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

AtomicBoolean 能防止一个进程重复提交任务，却不能协调三个同时运行的 Spring Boot 实例。每个实例都有自己的内存，每次重启还会丢失正在执行的任务标记。

本章继续使用 Java 21、Spring Boot 4.1.1 和已有 PostgreSQL。把待执行检查保存成数据库任务，用短事务认领、带令牌的租约和结果唯一约束，让实例退出后其他实例可以接手，并让失去所有权的旧实例无法提交结果。

1. 先确定可以承诺什么

为每个服务按数据库时间划分 30 秒调度窗口，同一 (service_id, scheduled_slot) 只创建一项业务任务。任务可以有多次执行尝试，但成功结果最多写入一次。

docs/monitoring-job-lifecycle.txt

1READY -> RUNNING(owner=A, lease=30s) -> DONE
2              |
3              +-- A 退出，租约到期 -> RUNNING(owner=B) -> DONE
4              +-- 可重试失败 -> READY(稍后重试)
5              +-- 达到 3 次尝试 -> FAILED

租约不可能撤回已经发出的 HTTP 请求。A 请求超时后，B 可能再次探测目标；A 甚至可能在 B 开始后才恢复。这里采用可重复执行的 at-least-once 处理思路，加上有限重试和单次持久化效果，不是 exactly-once 网络调用，也不保证失败任务最终一定成功。

探测使用只读 GET，重复调用通常可以接受。若远端操作有扣款等副作用，远端也必须接受稳定幂等键；仅给本地表加唯一索引不能让远端副作用自动幂等。

2. 创建任务与结果表

V6 已用于资源成员关系，V7 用于告警 Outbox，本章使用 V8。任务保存目标地址的配置快照，结果表使用 job_id 主键约束同一任务最多产生一份结果。

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

01create table monitoring_job (
02    id bigint generated always as identity primary key,
03    service_id bigint not null references monitored_service(id) on delete restrict,
04    service_name varchar(50) not null,
05    scheduled_slot bigint not null,
06    target_url text not null,
07    state varchar(16) not null default 'READY'
08        check (state in ('READY', 'RUNNING', 'DONE', 'FAILED')),
09    attempts integer not null default 0 check (attempts between 0 and 3),
10    available_at timestamptz not null default clock_timestamp(),
11    lease_until timestamptz,
12    owner_token uuid,
13    created_at timestamptz not null default clock_timestamp(),
14    unique (service_id, scheduled_slot),
15    check (
16        (state = 'RUNNING' and owner_token is not null and lease_until is not null)
17        or (state <> 'RUNNING' and owner_token is null and lease_until is null)
18    )
19);
20
21create index idx_monitoring_job_ready on monitoring_job (available_at, id)
22    where state = 'READY';
23create index idx_monitoring_job_expired on monitoring_job (lease_until, id)
24    where state = 'RUNNING';
25
26create table monitoring_job_result (
27    job_id bigint primary key references monitoring_job(id) on delete restrict,
28    status varchar(8) not null check (status in ('UP', 'DOWN')),
29    observed_at timestamptz not null default clock_timestamp()
30);

scheduled_slot 是业务标识的一部分，owner_token 是每一次认领的新 UUID，两者不能混用。重试保持相同 job ID，但必须换 token。这里只由受控配置生成 target_url，不能开放一个接收任意 URL 的匿名任务提交接口。

3. 在短事务中认领任务

下面是完整的数据库任务存储服务。每个公开方法都由另一个 Bean 调用，事务代理负责提交；HTTP 探测不放进这些事务中。

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

01package cn.usehook.spring.monitoring;
02
03import cn.usehook.spring.health.HealthState;
04import java.net.URI;
05import java.util.Optional;
06import java.util.UUID;
07import org.springframework.jdbc.core.simple.JdbcClient;
08import org.springframework.stereotype.Service;
09import org.springframework.transaction.annotation.Transactional;
10
11@Service
12public class MonitoringJobStore {
13    private final JdbcClient jdbc;
14
15    public MonitoringJobStore(JdbcClient jdbc) {
16        this.jdbc = jdbc;
17    }
18
19    public record Job(long id, String serviceName, URI targetUrl, UUID ownerToken) {}
20
21    @Transactional(timeout = 5)
22    public void enqueue(String serviceName, URI targetUrl) {
23        jdbc.sql("""
24            insert into monitoring_job
25                (service_id, service_name, scheduled_slot, target_url)
26            select id, name,
27                floor(extract(epoch from clock_timestamp()) / 30)::bigint, :url
28            from monitored_service where name = :service and enabled = true
29            on conflict (service_id, scheduled_slot) do nothing
30            """).param("service", serviceName).param("url", targetUrl.toString()).update();
31    }
32
33    @Transactional(timeout = 5)
34    public Optional<Job> claim() {
35        jdbc.sql("""
36            update monitoring_job
37            set state = 'FAILED', owner_token = null, lease_until = null
38            where state = 'RUNNING' and attempts >= 3
39              and lease_until <= clock_timestamp()
40            """).update();
41
42        return jdbc.sql("""
43            with candidate as (
44                select id from monitoring_job
45                where attempts < 3 and (
46                    (state = 'READY' and available_at <= clock_timestamp())
47                    or (state = 'RUNNING' and lease_until <= clock_timestamp())
48                )
49                order by available_at, id
50                limit 1
51                for update skip locked
52            )
53            update monitoring_job j
54            set state = 'RUNNING', owner_token = :token,
55                lease_until = clock_timestamp() + interval '30 seconds',
56                attempts = j.attempts + 1
57            from candidate c where j.id = c.id
58            returning j.id, j.service_name, j.target_url, j.owner_token
59            """).param("token", UUID.randomUUID())
60            .query((rs, row) -> new Job(rs.getLong("id"),
61                rs.getString("service_name"), URI.create(rs.getString("target_url")),
62                rs.getObject("owner_token", UUID.class))).optional();
63    }
64
65    @Transactional(timeout = 5)
66    public boolean complete(Job job, HealthState state) {
67        int accepted = jdbc.sql("""
68            update monitoring_job
69            set state = 'DONE', owner_token = null, lease_until = null
70            where id = :id and state = 'RUNNING' and owner_token = :token
71              and lease_until > clock_timestamp()
72            """).param("id", job.id()).param("token", job.ownerToken()).update();
73        if (accepted == 0) {
74            return false;
75        }
76
77        jdbc.sql("""
78            insert into monitoring_job_result(job_id, status)
79            values (:id, :status)
80            """).param("id", job.id()).param("status", state.name()).update();
81        return true;
82    }
83
84    @Transactional(timeout = 5)
85    public void retry(Job job) {
86        jdbc.sql("""
87            update monitoring_job
88            set state = case when attempts < 3 then 'READY' else 'FAILED' end,
89                available_at = clock_timestamp()
90                    + least(60, 5 * power(2, attempts - 1)) * interval '1 second',
91                owner_token = null, lease_until = null
92            where id = :id and state = 'RUNNING' and owner_token = :token
93              and lease_until > clock_timestamp()
94            """).param("id", job.id()).param("token", job.ownerToken()).update();
95    }
96}

生产者重复执行 INSERT 时，由数据库唯一约束完成竞争仲裁，不需要先查询“任务存在吗”。ON CONFLICT DO NOTHING 保留首次提交的地址快照，因此同一部署的副本必须使用一致目标配置。30 秒窗口由数据库计算，避免各实例机器时间不同导致业务键漂移。应用全部离线期间不会补造过去的窗口；需要补偿扫描时应另设有限回放范围。

认领过程将行锁和所有权更新放在同一条语句中。其他工作实例跳过被锁住的行，可继续处理别的任务；SKIP LOCKED 适合这样的队列表，不适合需要完整一致结果的普通业务查询。参见 PostgreSQL SELECT 锁定规则。

完成时先判断 job ID、状态、token 和租约，再写结果；两个操作处于同一事务，插入失败会撤销 DONE。判断成功的瞬间获得该行写锁，接管者会等待并在提交后看到 DONE。即使调用方没有收到提交成功响应，重新调用 complete 也不会增加第二份结果。

检查租约使用数据库 clock_timestamp()。current_timestamp 固定为事务开始时间，不适合表达等待之后的实际过期判断，区别见 PostgreSQL 时间函数。这个 token 是数据库内的所有权凭证，不是可以让任意外部系统自动拒绝旧请求的通用 fencing token。

4. 替换原来的调度器

移除旧 HealthCheckScheduler 或 AsyncHealthCheckScheduler 的 @Component，只保留一种调度入口；原来的单元测试仍可直接创建那些普通 Java 对象。否则旧调度器会继续绕过任务表访问目标并写入旧结果。

保留 @EnableScheduling，将新任务功能做成显式开关，方便测试禁用后台轮询：

src/main/resources/application.yaml

01app:
02  monitoring:
03    target-name: inventory
04    target-url: ${TARGET_HEALTH_URL:http://localhost:8081/api/v1/health}
05    jobs:
06      enabled: true
07spring:
08  task:
09    scheduling:
10      pool:
11        size: 2
12  http:
13    clients:
14      connect-timeout: 2s
15      read-timeout: 3s
16      redirects: dont-follow

MonitoringProperties 和 RemoteHealthClient 使用前文版本。目标名称需要在 monitored_service 中存在且 enabled 为 true；可复用资源授权篇的 inventory 教学数据。禁用服务会停止创建新任务，已经提交的任务仍按快照完成；若需要即时取消，必须另外增加取消状态和完成条件。

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

01package cn.usehook.spring.monitoring;
02
03import cn.usehook.spring.config.MonitoringProperties;
04import org.slf4j.Logger;
05import org.slf4j.LoggerFactory;
06import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
07import org.springframework.scheduling.annotation.Scheduled;
08import org.springframework.stereotype.Component;
09
10@Component
11@ConditionalOnProperty(prefix = "app.monitoring.jobs", name = "enabled",
12    havingValue = "true")
13public class DatabaseHealthScheduler {
14    private static final Logger log = LoggerFactory.getLogger(DatabaseHealthScheduler.class);
15    private final MonitoringJobStore jobs;
16    private final RemoteHealthClient client;
17    private final MonitoringProperties properties;
18
19    public DatabaseHealthScheduler(MonitoringJobStore jobs, RemoteHealthClient client,
20                                   MonitoringProperties properties) {
21        this.jobs = jobs;
22        this.client = client;
23        this.properties = properties;
24    }
25
26    @Scheduled(fixedDelay = 5000, initialDelay = 5000)
27    public void produce() {
28        jobs.enqueue(properties.targetName(), properties.targetUrl());
29    }
30
31    @Scheduled(fixedDelay = 1000, initialDelay = 1000)
32    public void consume() {
33        jobs.claim().ifPresent(job -> {
34            try {
35                ProbeResult result = client.probe(job.targetUrl());
36                if (!jobs.complete(job, result.state())) {
37                    log.warn("Discarded result from expired monitoring job owner: {}", job.id());
38                }
39            } catch (RuntimeException error) {
40                log.error("Monitoring job attempt failed: {}", job.id(), error);
41                jobs.retry(job);
42            }
43        });
44    }
45}

consume() 没有事务，claim 返回时认领事务已提交，网络等待期间没有持有数据库行锁或事务连接。一次只认领一个任务，每个实例最多一个同步消费调用；增加副本会增加并行消费者，不需要在本章再叠加 @Async 与无界任务队列。

RemoteHealthClient 已将超时、非成功响应等转换成 DOWN，因此 DOWN 是一次有效监控结果，会正常完成任务，不会立刻重试到变成 UP。retry 处理的是未能完成本次任务的应用异常。若数据库完全不可用、retry 本身也失败，记录日志后让租约过期，其他实例之后再接管；达到三次尝试的任务进入 FAILED，交给运维检查。

30 秒租约只是本案例预算，不等于保证所有 HTTP 调用都在 30 秒内结束。必须结合真实最坏耗时设置，过期结果会被丢弃。长任务若需要续租，续租同样必须附带 token 和未过期条件，不能让旧 owner 无条件延长新 owner 的租约。

5. 验证重复提交与旧 owner

下面的测试复用 PostgreSQL Testcontainers。特意不加测试类级 @Transactional，让每一次认领和完成都真正提交，避免一个测试事务掩盖真实租约边界。

src/test/java/cn/usehook/spring/monitoring/MonitoringJobTests.java

01package cn.usehook.spring.monitoring;
02
03import cn.usehook.spring.TestcontainersConfiguration;
04import cn.usehook.spring.health.HealthState;
05import java.util.UUID;
06import org.junit.jupiter.api.Test;
07import org.springframework.beans.factory.annotation.Autowired;
08import org.springframework.boot.test.context.SpringBootTest;
09import org.springframework.context.annotation.Import;
10import org.springframework.jdbc.core.simple.JdbcClient;
11import org.springframework.security.oauth2.jwt.JwtDecoder;
12import org.springframework.test.context.bean.override.mockito.MockitoBean;
13import static org.assertj.core.api.Assertions.assertThat;
14
15@SpringBootTest(properties = {
16    "app.monitoring.jobs.enabled=false",
17    "app.monitoring.initial-delay=1h",
18    "app.alerts.enabled=false",
19    "app.history.retention.enabled=false"
20})
21@Import(TestcontainersConfiguration.class)
22class MonitoringJobTests {
23    @Autowired MonitoringJobStore jobs;
24    @Autowired JdbcClient jdbc;
25    @MockitoBean JwtDecoder decoder;
26
27    @Test
28    void rejectsExpiredOwnerAndPersistsOneResult() {
29        String name = "job-" + UUID.randomUUID();
30        long group = jdbc.sql("insert into monitoring_group(name) values (:name) returning id")
31            .param("name", name).query(Long.class).single();
32        long service = jdbc.sql("""
33            insert into monitored_service(name, group_id) values (:name, :group) returning id
34            """).param("name", name).param("group", group).query(Long.class).single();
35        try {
36            // 固定窗口标识，避免测试刚好跨越真实的 30 秒边界。
37            for (int i = 0; i < 2; i++) {
38                jdbc.sql("""
39                    insert into monitoring_job(service_id, service_name, scheduled_slot, target_url)
40                    values (:service, :name, 123, 'http://inventory/api/v1/health')
41                    on conflict (service_id, scheduled_slot) do nothing
42                    """).param("service", service).param("name", name).update();
43            }
44            assertThat(jdbc.sql("select count(*) from monitoring_job where service_id=:id")
45                .param("id", service).query(Long.class).single()).isEqualTo(1L);
46            MonitoringJobStore.Job first = jobs.claim().orElseThrow();
47            assertThat(jobs.claim()).isEmpty();
48            jdbc.sql("update monitoring_job set lease_until=clock_timestamp()-interval '1 second' where id=:id")
49                .param("id", first.id()).update();
50            assertThat(jobs.complete(first, HealthState.UP)).isFalse();
51            MonitoringJobStore.Job second = jobs.claim().orElseThrow();
52            assertThat(second.id()).isEqualTo(first.id());
53            assertThat(second.ownerToken()).isNotEqualTo(first.ownerToken());
54            assertThat(jobs.complete(first, HealthState.UP)).isFalse();
55            assertThat(jobs.complete(second, HealthState.DOWN)).isTrue();
56            assertThat(jobs.complete(second, HealthState.DOWN)).isFalse();
57            assertThat(jdbc.sql("select status from monitoring_job_result where job_id=:id")
58                .param("id", second.id()).query(String.class).single()).isEqualTo("DOWN");
59        } finally {
60            jdbc.sql("delete from monitoring_job_result where job_id in (select id from monitoring_job where service_id=:id)")
61                .param("id", service).update();
62            jdbc.sql("delete from monitoring_job where service_id=:id").param("id", service).update();
63            jdbc.sql("delete from monitored_service where id=:id").param("id", service).update();
64            jdbc.sql("delete from monitoring_group where id=:id").param("id", group).update();
65        }
66    }
67}

执行 ./mvnw -Dtest=MonitoringJobTests test。测试使用专用空容器任务表，不允许并行用例或其他后台消费者写同一张队列表；清理条件只针对本次生成的服务 ID。通过主动调整测试租约制造过期，不需要等待 30 秒或依靠不稳定的 sleep。

再补三组故障验证：两个独立连接同时 claim 不能得到同一有效任务；第三次尝试仍失败时进入 FAILED；complete 的结果插入失败时，DONE 必须随事务回滚。它们分别验证并发认领、重试上限和原子提交，不能只断言 Java 方法调用次数。

6. 观察两实例接管

在共享教学 PostgreSQL 上启动两个实例，端口分别设为 8080、8082，目标配置相同。不要启动旧调度器。观察任务状态与结果数量：

scripts/inspect-monitoring-jobs.sql

01select j.id, j.service_id, j.scheduled_slot, j.state, j.attempts,
02       j.owner_token, j.lease_until, r.status
03from monitoring_job j
04left join monitoring_job_result r on r.job_id = j.id
05order by j.id desc
06limit 20;
07
08select service_id, scheduled_slot, count(*)
09from monitoring_job
10group by service_id, scheduled_slot
11having count(*) > 1;

正常情况下第二条查询为空，第一条中的大多数任务一次尝试后变为 DONE。若需要演示接管，在专用调试环境让 A 停在“claim 已返回、complete 尚未调用”的断点；租约过期后 B 会使用新 token 认领。恢复 A 时 complete 必须返回 false。不要在生产环境通过暂停所有线程模拟这个过程。

检查延迟超过一个调度窗口时，不同窗口仍可能同时执行，这是任务之间的正常并行；本方案只防止同一有效租约被重复认领，不承诺一个服务永远只有一条远程请求在途。所有实例离线、数据库长时间故障或三次尝试耗尽时，需要运维介入，不能把 FAILED 自动无上限重放。

7. 与历史和告警衔接

本章结果保存在 monitoring_job_result，原来的手动检查入口仍写已有历史表。消费者故意只调用 RemoteHealthClient，不调用会提前落库的 HealthCheckService；否则旧 owner 可能先写历史、发告警，最后才发现 complete 失败。

若要接入前面的状态历史和 Outbox，应把它们的数据库写入移动到 complete 的 accepted 检查之后、同一个事务之内，并用 job ID 作为事件幂等标识。还要处理不同 scheduled_slot 乱序完成：旧窗口结果可以追加历史，但不能覆盖较新窗口的“当前状态”。最直接的办法是在最新状态表保存最后处理的 slot，并在 UPDATE 中要求新 slot 更大；不能直接复用无条件覆盖的旧 Repository。

任务和结果表也需要保留策略。清理 DONE/FAILED 任务前先清理关联结果，并明确去重窗口：唯一记录一旦删除，就不再阻止相同业务键重新创建。当前生产者只创建当前窗口，不回放过去，因此历史清理与日常调度不会冲突；增加人工重放功能时必须重新设计这一约束。

PostgreSQL 已经提供本案例所需的唯一约束、条件更新和行锁，暂时无需引入额外分布式锁服务。数据库队列成为瓶颈、需要大量独立消费者或跨系统投递时，再评估消息系统。ON CONFLICT 的原子行为见 INSERT 官方文档，Spring 方法事务的边界见事务传播官方文档。

首页订阅