优化
This commit is contained in:
+3
-6
@@ -39,12 +39,8 @@ public class CarrierScheduleQueryScheduler {
|
||||
this.repository = repository; this.service = service; this.integration = integration;
|
||||
}
|
||||
|
||||
@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}")
|
||||
@Scheduled(fixedDelayString = "${baiyue.carrier-schedule.scheduler.scan-fixed-delay:PT1M}",
|
||||
initialDelayString = "${baiyue.carrier-schedule.scheduler.scan-initial-delay:PT10S}")
|
||||
public void poll() {
|
||||
LocalDateTime triggerAt = LocalDateTime.ofInstant(clock.instant(), ZoneId.of(scheduleZone));
|
||||
int batch = Math.max(1, Math.min(batchSize, 100));
|
||||
@@ -65,6 +61,7 @@ public class CarrierScheduleQueryScheduler {
|
||||
/** Executes only after this worker wins the database lease. */
|
||||
void execute(CarrierScheduleQueryEntity candidate, LocalDateTime now) {
|
||||
if (!repository.claimLease(candidate.getId(), workerId, now, now.plus(leaseDuration))) return;
|
||||
if (service.deferMissedNormalSlot(candidate.getId(), now)) return;
|
||||
try {
|
||||
var response = integration.query(internalToken,
|
||||
new QueryRequest(candidate.getCarrierCode(), candidate.getTrackingType(), candidate.getTrackingNo()));
|
||||
|
||||
+55
-11
@@ -53,6 +53,7 @@ import java.util.Set;
|
||||
|
||||
@Service
|
||||
public class CarrierScheduleQueryService {
|
||||
static final int MAX_SLOT_ATTEMPTS = 3;
|
||||
private static final Set<String> TRACKING_TYPES = Set.of("BOOKING_NO", "BILL_OF_LADING", "CONTAINER_NO");
|
||||
|
||||
@Resource private CarrierScheduleQueryRepository repository;
|
||||
@@ -217,34 +218,76 @@ public class CarrierScheduleQueryService {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public boolean deferMissedNormalSlot(long id, LocalDateTime scanAt) {
|
||||
CarrierScheduleQueryEntity value = requiredLocked(id);
|
||||
if ("STOPPED".equals(value.getStatus())) return false;
|
||||
int attempts = value.getAttemptCount() == null ? 0 : value.getAttemptCount();
|
||||
boolean staleRetry = attempts > 0 && value.getWindowStartedAt() != null
|
||||
&& !scanAt.isBefore(CarrierScheduleQuerySlots.nextSlot(value.getWindowStartedAt()));
|
||||
if (staleRetry) {
|
||||
LocalDateTime currentSlot = CarrierScheduleQuerySlots.fixedSlotInCurrentMinute(scanAt);
|
||||
if (currentSlot != null) {
|
||||
value.setNextAllowedAt(currentSlot); value.setWindowStartedAt(currentSlot);
|
||||
value.setAttemptCount(0); value.setConsecutiveFailures(0);
|
||||
if (repository.updateState(value, value.getVersion(), 0L, "carrier-scheduler") != 1) throw concurrent();
|
||||
return false;
|
||||
}
|
||||
} else if (!CarrierScheduleQuerySlots.isMissedNormalSlot(
|
||||
value.getNextAllowedAt(), value.getAttemptCount(), scanAt)) return false;
|
||||
value.setLeaseOwner(null); value.setLeaseUntil(null);
|
||||
value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(scanAt));
|
||||
value.setWindowStartedAt(value.getNextAllowedAt());
|
||||
value.setAttemptCount(0); value.setConsecutiveFailures(0);
|
||||
if (repository.updateState(value, value.getVersion(), 0L, "carrier-scheduler") != 1) throw concurrent();
|
||||
return true;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void completeScheduledAttempt(long id, String outcome, String errorCode, String errorMessage) {
|
||||
CarrierScheduleQueryEntity value = requiredLocked(id);
|
||||
insertAttempt(value, outcome, errorCode, errorMessage);
|
||||
int slotAttemptNo = Math.min(MAX_SLOT_ATTEMPTS, increment(value.getAttemptCount()));
|
||||
insertAttempt(value, slotAttemptNo, outcome, errorCode, errorMessage);
|
||||
if ("STOPPED".equals(value.getStatus())) return;
|
||||
CarrierScheduleQueryEntity before = copy(value);
|
||||
LocalDateTime now = now();
|
||||
value.setLeaseOwner(null); value.setLeaseUntil(null); value.setLastAttemptAt(now);
|
||||
value.setLastAttemptStatus(outcome); value.setLastErrorCode(trim(errorCode));
|
||||
value.setLastErrorMessage(trim(errorMessage));
|
||||
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));
|
||||
LocalDateTime nextFixedSlot = CarrierScheduleQuerySlots.nextSlot(now);
|
||||
if ("RETRYABLE_FAILURE".equals(outcome)) {
|
||||
value.setStatus("READY");
|
||||
value.setConsecutiveFailures(slotAttemptNo);
|
||||
LocalDateTime retryAt = now.plusMinutes(15L << Math.max(0, slotAttemptNo - 1));
|
||||
if (slotAttemptNo < MAX_SLOT_ATTEMPTS && retryAt.isBefore(nextFixedSlot)) {
|
||||
if (slotAttemptNo == 1 || value.getWindowStartedAt() == null) {
|
||||
value.setWindowStartedAt(value.getNextAllowedAt());
|
||||
}
|
||||
value.setAttemptCount(slotAttemptNo); value.setNextAllowedAt(retryAt);
|
||||
} else {
|
||||
resetForNextFixedSlot(value, nextFixedSlot);
|
||||
}
|
||||
} else {
|
||||
value.setStatus("MANUAL_REQUIRED".equals(outcome) || "PERMANENT_FAILURE".equals(outcome)
|
||||
? "MANUAL_REQUIRED" : "READY");
|
||||
resetForNextFixedSlot(value, nextFixedSlot);
|
||||
}
|
||||
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,
|
||||
"carrier-scheduler");
|
||||
}
|
||||
|
||||
private void insertAttempt(CarrierScheduleQueryEntity value, String outcome,
|
||||
private static void resetForNextFixedSlot(CarrierScheduleQueryEntity value, LocalDateTime nextFixedSlot) {
|
||||
value.setConsecutiveFailures(0); value.setAttemptCount(0);
|
||||
value.setWindowStartedAt(nextFixedSlot); value.setNextAllowedAt(nextFixedSlot);
|
||||
}
|
||||
|
||||
private void insertAttempt(CarrierScheduleQueryEntity value, int slotAttemptNo, String outcome,
|
||||
String errorCode, String errorMessage) {
|
||||
CarrierScheduleQueryAttemptEntity attempt = new CarrierScheduleQueryAttemptEntity();
|
||||
attempt.setId(IdWorker.getId()); attempt.setQueryId(value.getId());
|
||||
attempt.setWindowAttemptNo(increment(value.getAttemptCount()));
|
||||
attempt.setWindowAttemptNo(slotAttemptNo);
|
||||
attempt.setCarrierCode(value.getCarrierCode()); attempt.setAccessMode(value.getAccessMode());
|
||||
attempt.setOfficialUrl(value.getOfficialUrl()); attempt.setAttemptStatus(outcome);
|
||||
boolean manual = "MANUAL_REQUIRED".equals(outcome);
|
||||
@@ -301,7 +344,8 @@ public class CarrierScheduleQueryService {
|
||||
value.setStatus("READY"); value.setResultPortScheduleId(schedule.getId());
|
||||
value.setLastSuccessAt(now); value.setSourceUpdatedAt(schedule.getObservedAt());
|
||||
value.setExpiresAt(schedule.getExpiresAt()); value.setNextAllowedAt(CarrierScheduleQuerySlots.nextSlot(now));
|
||||
value.setWindowStartedAt(now); value.setLastAttemptStatus("MANUAL_COMPLETED");
|
||||
value.setWindowStartedAt(value.getNextAllowedAt()); value.setAttemptCount(0);
|
||||
value.setConsecutiveFailures(0); 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();
|
||||
CarrierScheduleQueryEntity after = required(id);
|
||||
|
||||
+24
@@ -17,4 +17,28 @@ public final class CarrierScheduleQuerySlots {
|
||||
if (now.toLocalTime().isBefore(AFTERNOON)) return now.toLocalDate().atTime(AFTERNOON);
|
||||
return now.toLocalDate().plusDays(1).atTime(MORNING);
|
||||
}
|
||||
|
||||
public static boolean isFixedSlot(LocalDateTime value) {
|
||||
if (value == null || value.getSecond() != 0 || value.getNano() != 0) return false;
|
||||
return value.toLocalTime().equals(MORNING) || value.toLocalTime().equals(NOON)
|
||||
|| value.toLocalTime().equals(AFTERNOON);
|
||||
}
|
||||
|
||||
public static boolean isWithinFixedSlotMinute(LocalDateTime value) {
|
||||
if (value == null) return false;
|
||||
LocalTime minute = value.toLocalTime().withSecond(0).withNano(0);
|
||||
return minute.equals(MORNING) || minute.equals(NOON) || minute.equals(AFTERNOON);
|
||||
}
|
||||
|
||||
public static LocalDateTime fixedSlotInCurrentMinute(LocalDateTime value) {
|
||||
return isWithinFixedSlotMinute(value) ? value.withSecond(0).withNano(0) : null;
|
||||
}
|
||||
|
||||
/** A normal slot is executable only during its scheduled minute; technical retry times are unaffected. */
|
||||
public static boolean isMissedNormalSlot(LocalDateTime nextAllowedAt, Integer attemptCount,
|
||||
LocalDateTime scanAt) {
|
||||
return (attemptCount == null || attemptCount == 0) && isFixedSlot(nextAllowedAt)
|
||||
&& !isWithinFixedSlotMinute(scanAt)
|
||||
&& !scanAt.isBefore(nextAllowedAt.plusMinutes(1));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,9 +35,8 @@ baiyue:
|
||||
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 * * *}
|
||||
scan-fixed-delay: ${CARRIER_SCHEDULE_SCAN_FIXED_DELAY:PT1M}
|
||||
scan-initial-delay: ${CARRIER_SCHEDULE_SCAN_INITIAL_DELAY:PT10S}
|
||||
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}
|
||||
|
||||
@@ -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 '累计自动查询调用次数',
|
||||
next_allowed_at DATETIME(3) NULL COMMENT '下一个09:00、12:30或16:00固定查询槽位;停止后为空',
|
||||
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 '下一正常槽位或槽内技术退避时间;停止后为空',
|
||||
last_attempt_at DATETIME(3) NULL COMMENT '最近一次查询尝试时间',
|
||||
last_success_at DATETIME(3) NULL COMMENT '最近一次取得有效结果时间',
|
||||
source_updated_at DATETIME(3) NULL COMMENT '船公司官网展示的结果更新时间',
|
||||
@@ -64,13 +64,14 @@ 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'))
|
||||
) ENGINE=InnoDB COMMENT='六家船公司固定时刻自动查询任务';
|
||||
'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='六家船公司固定槽位与技术退避查询任务';
|
||||
|
||||
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 '任务累计自动查询调用序号',
|
||||
window_attempt_no INT UNSIGNED NOT NULL COMMENT '当前正常查询槽位内的调用序号;取值1至3',
|
||||
carrier_code VARCHAR(16) NOT NULL COMMENT '本次尝试的船公司编码快照',
|
||||
access_mode VARCHAR(32) NOT NULL COMMENT '本次尝试的官网访问方式快照',
|
||||
official_url VARCHAR(500) NOT NULL COMMENT '本次尝试使用的官网地址快照',
|
||||
@@ -86,6 +87,7 @@ 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='船司官网查询尝试不可变审计';
|
||||
|
||||
@@ -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 and freshness fields; automatic execution uses the fixed Shanghai slots}
|
||||
'200': {description: Query tasks with source and freshness fields; normal execution uses fixed Shanghai slots while retryable technical failures may retry within the current slot}
|
||||
post:
|
||||
summary: Create a query task linked to a real booking and allocated container
|
||||
x-permission: platform:firstleg:port-schedule-query:create
|
||||
|
||||
Reference in New Issue
Block a user