java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
ouyangxq
2026-08-21 16:15:02 +08:00
parent 8e07342fa1
commit cd477fe40a
16 changed files with 278 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
firstleg/src/main/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleQueryScheduler.java
+24 -3
View File
@@ -13,6 +13,7 @@ import org.springframework.stereotype.Component;
import java.time.Clock;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;
import java.util.UUID;
@@ -26,6 +27,9 @@ public class CarrierScheduleQueryScheduler {
@Value("${baiyue.security.internal-service-token:}") private String internalToken;
@Value("${baiyue.carrier-schedule.scheduler.batch-size:20}") private int batchSize = 20;
@Value("${baiyue.carrier-schedule.scheduler.max-tasks-per-trigger:2000}")
private int maxTasksPerTrigger = 2000;
@Value("${baiyue.carrier-schedule.scheduler.zone:Asia/Shanghai}") private String scheduleZone = "Asia/Shanghai";
@Value("${baiyue.carrier-schedule.scheduler.lease-duration:PT5M}")
private Duration leaseDuration = Duration.ofMinutes(5);
@@ -35,10 +39,27 @@ public class CarrierScheduleQueryScheduler {
this.repository = repository; this.service = service; this.integration = integration;
}
@Scheduled(fixedDelayString = "${baiyue.carrier-schedule.scheduler.fixed-delay:PT1M}")
@Scheduled(cron = "${baiyue.carrier-schedule.scheduler.cron-morning:0 0 9 * * *}",
zone = "${baiyue.carrier-schedule.scheduler.zone:Asia/Shanghai}")
@Scheduled(cron = "${baiyue.carrier-schedule.scheduler.cron-noon:0 30 12 * * *}",
zone = "${baiyue.carrier-schedule.scheduler.zone:Asia/Shanghai}")
@Scheduled(cron = "${baiyue.carrier-schedule.scheduler.cron-afternoon:0 0 16 * * *}",
zone = "${baiyue.carrier-schedule.scheduler.zone:Asia/Shanghai}")
public void poll() {
LocalDateTime now = LocalDateTime.now(clock);
repository.due(now, Math.max(1, Math.min(batchSize, 100))).forEach(candidate -> execute(candidate, now));
LocalDateTime triggerAt = LocalDateTime.ofInstant(clock.instant(), ZoneId.of(scheduleZone));
int batch = Math.max(1, Math.min(batchSize, 100));
int safetyLimit = Math.max(1, maxTasksPerTrigger);
int scanned = 0;
while (scanned < safetyLimit) {
List<CarrierScheduleQueryEntity> candidates = repository.due(triggerAt,
Math.min(batch, safetyLimit - scanned));
if (candidates.isEmpty()) return;
for (CarrierScheduleQueryEntity candidate : candidates) {
execute(candidate, triggerAt); scanned++;
if (scanned >= safetyLimit) return;
}
if (candidates.size() < batch) return;
}
}
/** Executes only after this worker wins the database lease. */
firstleg/src/main/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleQueryService.java
+21 -34
View File
@@ -35,6 +35,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
@@ -43,6 +44,7 @@ import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Clock;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.HexFormat;
import java.util.List;
import java.util.Locale;
@@ -51,14 +53,14 @@ import java.util.Set;
@Service
public class CarrierScheduleQueryService {
static final int MAX_ATTEMPTS = 3;
static final int WINDOW_DAYS = 3;
private static final Set<String> TRACKING_TYPES = Set.of("BOOKING_NO", "BILL_OF_LADING", "CONTAINER_NO");
@Resource private CarrierScheduleQueryRepository repository;
@Resource private PortScheduleRepository portSchedules;
@Resource private CarrierScheduleAdapterRegistry adapters;
@Resource private ObjectMapper objectMapper;
@Value("${baiyue.carrier-schedule.scheduler.zone:Asia/Shanghai}")
private String scheduleZone = "Asia/Shanghai";
private Clock clock = Clock.systemDefaultZone();
public List<CarrierDescriptor> carriers() { return adapters.descriptors(); }
@@ -114,7 +116,8 @@ public class CarrierScheduleQueryService {
value.setStatus("READY"); value.setTriggerSource("OUTBOUND_CONFIRM");
value.setLatestStandardNode(CarrierScheduleNode.DEPARTED_WAREHOUSE.name());
value.setLatestNodeObservedAt(now); value.setWindowStartedAt(now); value.setAttemptCount(0);
value.setConsecutiveFailures(0); value.setNextAllowedAt(now); value.setCreatedBy(operatorId); value.setCreatedByName(operatorName);
value.setConsecutiveFailures(0); value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now));
value.setCreatedBy(operatorId); value.setCreatedByName(operatorName);
value.setUpdatedBy(operatorId); value.setUpdatedByName(operatorName); value.setVersion(0);
try { repository.insert(value); }
catch (DuplicateKeyException exception) {
@@ -206,7 +209,7 @@ public class CarrierScheduleQueryService {
value.setStatus("STOPPED"); value.setStoppedAt(now()); value.setStopReason("已还柜");
value.setLeaseOwner(null); value.setLeaseUntil(null); value.setNextAllowedAt(null);
} else {
value.setStatus("READY"); value.setNextAllowedAt(now().plusDays(WINDOW_DAYS));
value.setStatus("READY"); value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now()));
}
if (repository.updateState(value, value.getVersion(), operatorId, operatorName) != 1) throw concurrent();
return true;
@@ -224,31 +227,13 @@ public class CarrierScheduleQueryService {
value.setLeaseOwner(null); value.setLeaseUntil(null); value.setLastAttemptAt(now);
value.setLastAttemptStatus(outcome); value.setLastErrorCode(trim(errorCode));
value.setLastErrorMessage(trim(errorMessage));
if ("SUCCESS".equals(outcome) || "NO_CHANGE".equals(outcome)) {
value.setStatus("READY");
value.setConsecutiveFailures(0); value.setAttemptCount(0); value.setWindowStartedAt(now);
value.setNextAllowedAt(now.plusDays(WINDOW_DAYS));
} else {
int failures = count(value.getConsecutiveFailures()) + 1;
LocalDateTime windowStart = value.getWindowStartedAt();
if (windowStart == null || !now.isBefore(windowStart.plusDays(WINDOW_DAYS))) {
windowStart = now; value.setAttemptCount(0);
}
int attempts = count(value.getAttemptCount()) + 1;
value.setWindowStartedAt(windowStart); value.setAttemptCount(Math.min(MAX_ATTEMPTS, attempts));
value.setConsecutiveFailures(failures);
if ("MANUAL_REQUIRED".equals(outcome) || "PERMANENT_FAILURE".equals(outcome)) {
value.setStatus("MANUAL_REQUIRED");
value.setNextAllowedAt(now.plusDays(WINDOW_DAYS));
} else {
long minutes = Math.min(24L * 60L, 15L << Math.min(failures - 1, 6));
LocalDateTime retryAt = now.plusMinutes(minutes);
if (attempts >= MAX_ATTEMPTS && retryAt.isBefore(windowStart.plusDays(WINDOW_DAYS))) {
retryAt = windowStart.plusDays(WINDOW_DAYS);
}
value.setNextAllowedAt(retryAt);
}
}
boolean success = "SUCCESS".equals(outcome) || "NO_CHANGE".equals(outcome);
value.setStatus("MANUAL_REQUIRED".equals(outcome) || "PERMANENT_FAILURE".equals(outcome)
? "MANUAL_REQUIRED" : "READY");
value.setConsecutiveFailures(success ? 0 : increment(value.getConsecutiveFailures()));
value.setAttemptCount(increment(value.getAttemptCount()));
if (value.getWindowStartedAt() == null) value.setWindowStartedAt(now);
value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now));
if (repository.updateState(value, value.getVersion(), 0L, "carrier-scheduler") != 1) throw concurrent();
CarrierScheduleQueryEntity after = required(id);
addHistory(id, "PROVIDER_QUERY", before.getVersion(), after.getVersion(), outcome, before, after, 0L,
@@ -259,7 +244,7 @@ public class CarrierScheduleQueryService {
String errorCode, String errorMessage) {
CarrierScheduleQueryAttemptEntity attempt = new CarrierScheduleQueryAttemptEntity();
attempt.setId(IdWorker.getId()); attempt.setQueryId(value.getId());
attempt.setWindowAttemptNo(Math.max(1, Math.min(MAX_ATTEMPTS, count(value.getAttemptCount()) + 1)));
attempt.setWindowAttemptNo(increment(value.getAttemptCount()));
attempt.setCarrierCode(value.getCarrierCode()); attempt.setAccessMode(value.getAccessMode());
attempt.setOfficialUrl(value.getOfficialUrl()); attempt.setAttemptStatus(outcome);
boolean manual = "MANUAL_REQUIRED".equals(outcome);
@@ -287,7 +272,8 @@ public class CarrierScheduleQueryService {
value.setOfficialUrl(strategy.officialUrl()); value.setAccessMode(strategy.accessMode());
value.setStatus("READY"); value.setTriggerSource("MANUAL");
value.setWindowStartedAt(now); value.setAttemptCount(0); value.setConsecutiveFailures(0);
value.setNextAllowedAt(now); value.setManualReason(strategy.manualReason()); value.setRemark(trim(request.remark()));
value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now));
value.setManualReason(strategy.manualReason()); value.setRemark(trim(request.remark()));
value.setCreatedBy(operatorId); value.setCreatedByName(operatorName);
value.setUpdatedBy(operatorId); value.setUpdatedByName(operatorName); value.setVersion(0);
try { repository.insert(value); }
@@ -314,7 +300,7 @@ public class CarrierScheduleQueryService {
LocalDateTime now = now();
value.setStatus("READY"); value.setResultPortScheduleId(schedule.getId());
value.setLastSuccessAt(now); value.setSourceUpdatedAt(schedule.getObservedAt());
value.setExpiresAt(schedule.getExpiresAt()); value.setNextAllowedAt(now.plusDays(WINDOW_DAYS));
value.setExpiresAt(schedule.getExpiresAt()); value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now));
value.setWindowStartedAt(now); value.setLastAttemptStatus("MANUAL_COMPLETED");
value.setManualReason(trim(request.reason())); value.setLastErrorCode(null); value.setLastErrorMessage(null);
if (repository.updateState(value, request.expectedVersion(), operatorId, operatorName) != 1) throw concurrent();
@@ -505,8 +491,9 @@ public class CarrierScheduleQueryService {
return normalized;
}
private LocalDateTime now() { return LocalDateTime.now(clock); }
private static int count(Integer value) { return value == null ? 0 : value; }
private LocalDateTime now() { return LocalDateTime.ofInstant(clock.instant(), ZoneId.of(scheduleZone)); }
private static int increment(Integer value) { return value == null || value < 0 ? 1
: value == Integer.MAX_VALUE ? Integer.MAX_VALUE : value + 1; }
private static void validateKey(String key) {
if (!StringUtils.hasText(key) || key.length() > 128) throw badRequest("Idempotency-Key不能为空且不能超过128个字符");
}
firstleg/src/main/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleQuerySlots.java
+20
View File
@@ -0,0 +1,20 @@
package com.baiyue.firstleg.portschedule.service;
import java.time.LocalDateTime;
import java.time.LocalTime;
public final class CarrierScheduleQuerySlots {
private static final LocalTime MORNING = LocalTime.of(9, 0);
private static final LocalTime NOON = LocalTime.of(12, 30);
private static final LocalTime AFTERNOON = LocalTime.of(16, 0);
private CarrierScheduleQuerySlots() { }
/** Returns the first fixed slot strictly after {@code now}. Exact slot boundaries advance. */
public static LocalDateTime nextSlot(LocalDateTime now) {
if (now.toLocalTime().isBefore(MORNING)) return now.toLocalDate().atTime(MORNING);
if (now.toLocalTime().isBefore(NOON)) return now.toLocalDate().atTime(NOON);
if (now.toLocalTime().isBefore(AFTERNOON)) return now.toLocalDate().atTime(AFTERNOON);
return now.toLocalDate().plusDays(1).atTime(MORNING);
}
}
firstleg/src/main/resources/application.yml
+8
View File
@@ -33,6 +33,14 @@ baiyue:
cosco-official-api-enabled: ${COSCO_OPEN_API_ENABLED:false}
emc-automated: ${EMC_TRACKING_ENABLED:false}
hede-automated: ${HEDE_TRACKING_ENABLED:false}
scheduler:
zone: ${CARRIER_SCHEDULE_ZONE:Asia/Shanghai}
cron-morning: ${CARRIER_SCHEDULE_CRON_MORNING:0 0 9 * * *}
cron-noon: ${CARRIER_SCHEDULE_CRON_NOON:0 30 12 * * *}
cron-afternoon: ${CARRIER_SCHEDULE_CRON_AFTERNOON:0 0 16 * * *}
batch-size: ${CARRIER_SCHEDULE_BATCH_SIZE:20}
max-tasks-per-trigger: ${CARRIER_SCHEDULE_MAX_TASKS_PER_TRIGGER:2000}
lease-duration: ${CARRIER_SCHEDULE_LEASE_DURATION:PT5M}
security:
jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start}
internal-service-token: ${INTERNAL_SERVICE_TOKEN:change-me-to-an-internal-token-at-least-32-bytes-long}
firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql
+7 -9
View File
@@ -21,10 +21,10 @@ CREATE TABLE IF NOT EXISTS fl_carrier_schedule_query (
stop_reason VARCHAR(500) NULL COMMENT '任务停止原因',
lease_owner VARCHAR(64) NULL COMMENT '当前调度租约持有者',
lease_until DATETIME(3) NULL COMMENT '当前调度租约过期时间',
consecutive_failures INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '连续失败次数，用于指数退避',
window_started_at DATETIME(3) NOT NULL COMMENT '当前三天查询窗口起始时间',
attempt_count INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '当前查询窗口已执行次数；最多3次',
next_allowed_at DATETIME(3) NULL COMMENT '频控规则允许的下次查询时间；停止后为空',
consecutive_failures INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '连续失败次数诊断值；不改变固定查询槽位',
window_started_at DATETIME(3) NOT NULL COMMENT '首次进入自动查询调度的时间',
attempt_count INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '累计自动查询调用次数',
next_allowed_at DATETIME(3) NULL COMMENT '下一个09:00、12:30或16:00固定查询槽位；停止后为空',
last_attempt_at DATETIME(3) NULL COMMENT '最近一次查询尝试时间',
last_success_at DATETIME(3) NULL COMMENT '最近一次取得有效结果时间',
source_updated_at DATETIME(3) NULL COMMENT '船公司官网展示的结果更新时间',
@@ -64,14 +64,13 @@ CREATE TABLE IF NOT EXISTS fl_carrier_schedule_query (
CONSTRAINT chk_carrier_schedule_query_node CHECK
(latest_standard_node IS NULL OR latest_standard_node IN
('DEPARTED_WAREHOUSE','AT_ORIGIN_PORT','LOADED_ON_VESSEL','DEPARTED_ORIGIN_PORT','IN_TRANSIT',
'ARRIVED_DESTINATION_PORT','DISCHARGED_FROM_VESSEL','PICKUP_READY','PICKED_UP','EMPTY_RETURNED')),
CONSTRAINT chk_carrier_schedule_query_attempt CHECK (attempt_count BETWEEN 0 AND 3)
) ENGINE=InnoDB COMMENT='六家船公司官网查询任务与三天窗口频控';
'ARRIVED_DESTINATION_PORT','DISCHARGED_FROM_VESSEL','PICKUP_READY','PICKED_UP','EMPTY_RETURNED'))
) ENGINE=InnoDB COMMENT='六家船公司固定时刻自动查询任务';
CREATE TABLE IF NOT EXISTS fl_carrier_schedule_query_attempt (
id BIGINT UNSIGNED NOT NULL COMMENT '船司官网查询尝试ID',
query_id BIGINT UNSIGNED NOT NULL COMMENT '所属船司官网查询任务ID',
window_attempt_no INT UNSIGNED NOT NULL COMMENT '当前三天窗口内的尝试序号；取值1至3',
window_attempt_no INT UNSIGNED NOT NULL COMMENT '任务累计自动查询调用序号',
carrier_code VARCHAR(16) NOT NULL COMMENT '本次尝试的船公司编码快照',
access_mode VARCHAR(32) NOT NULL COMMENT '本次尝试的官网访问方式快照',
official_url VARCHAR(500) NOT NULL COMMENT '本次尝试使用的官网地址快照',
@@ -87,7 +86,6 @@ CREATE TABLE IF NOT EXISTS fl_carrier_schedule_query_attempt (
PRIMARY KEY (id),
KEY idx_carrier_schedule_attempt_query (query_id,created_at,id),
CONSTRAINT fk_carrier_schedule_attempt_query FOREIGN KEY (query_id) REFERENCES fl_carrier_schedule_query(id),
CONSTRAINT chk_carrier_schedule_attempt_no CHECK (window_attempt_no BETWEEN 1 AND 3),
CONSTRAINT chk_carrier_schedule_attempt_status CHECK
(attempt_status IN ('SUCCESS','NO_CHANGE','RETRYABLE_FAILURE','PERMANENT_FAILURE','MANUAL_REQUIRED'))
) ENGINE=InnoDB COMMENT='船司官网查询尝试不可变审计';
firstleg/src/main/resources/openapi/carrier-schedule-query-v1.yaml
+1 -1
View File
@@ -25,7 +25,7 @@ paths:
- {name: page, in: query, schema: {type: integer, minimum: 1, default: 1}}
- {name: size, in: query, schema: {type: integer, minimum: 1, maximum: 200, default: 20}}
responses:
'200': {description: Query tasks with source, freshness and rate-limit fields}
'200': {description: Query tasks with source and freshness fields; automatic execution uses the fixed Shanghai slots}
post:
summary: Create a query task linked to a real booking and allocated container
x-permission: platform:firstleg:port-schedule-query:create