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-09-29 17:55:54 +08:00
parent 41d8387cd5
commit 54ad26c607
111 changed files with 1837 additions and 962 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java
+28
View File
@@ -0,0 +1,28 @@
package com.baiyue.lastleg.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventTypes;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
/** Dispatches each supported business event directly to its owning service. */
@Component
@RequiredArgsConstructor
public class DirectEventClient {
private final FirstlegEventClient firstleg;
private final FinanceBusinessEventClient finance;
public ApiResponse<DeliveryAck> receive(String token, EventEnvelope event) {
event.requireValid();
return switch (event.eventType()) {
case EventTypes.OVERSEAS_CONTAINER_UNLOADED,
EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED,
EventTypes.OVERSEAS_DELIVERY_COMPLETED -> firstleg.receive(token, event);
case EventTypes.WALLET_COMMAND_REQUESTED -> finance.wallet(token, event);
case EventTypes.BUSINESS_FEE_SOURCE_CHANGED -> finance.fee(token, event);
default -> throw new IllegalArgumentException("Unsupported outbound event type: " + event.eventType());
};
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/client/EventBridgeClient.java
-20
View File
@@ -1,20 +0,0 @@
package com.baiyue.lastleg.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.EventEnvelope;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
/** 将海外业务 Outbox 投递到国内区域事件桥。 */
@FeignClient(name = "event-bridge-service", contextId = "overseasContainerEventBridgeClient",
url = "${CN_EVENT_BRIDGE_URL:${baiyue.firstleg.event-bridge-url:}}")
public interface EventBridgeClient {
@PostMapping("/internal/event-bridge/events")
ApiResponse<Acknowledgement> receive(
@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody EventEnvelope envelope);
record Acknowledgement(String eventId, String result, String receivedAt) { }
}
lastleg-service/src/main/java/com/baiyue/lastleg/client/FinanceBusinessEventClient.java
+21
View File
@@ -0,0 +1,21 @@
package com.baiyue.lastleg.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.web.EventDeliveryFeignConfiguration;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelope;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
@FeignClient(configuration = EventDeliveryFeignConfiguration.class, name = "finance-service", contextId = "FinanceBusinessEventClient",
url = "${FINANCE_SERVICE_URL:${baiyue.clients.finance.url:}}")
public interface FinanceBusinessEventClient {
@PostMapping("/internal/finance/v1/wallet-command-events")
ApiResponse<DeliveryAck> wallet(@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody EventEnvelope envelope);
@PostMapping("/internal/finance/v1/business-fee-source-events")
ApiResponse<DeliveryAck> fee(@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody EventEnvelope envelope);
}
lastleg-service/src/main/java/com/baiyue/lastleg/client/FirstlegEventClient.java
+18
View File
@@ -0,0 +1,18 @@
package com.baiyue.lastleg.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.web.EventDeliveryFeignConfiguration;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelope;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
@FeignClient(configuration = EventDeliveryFeignConfiguration.class, name = "firstleg-service", contextId = "FirstlegEventClient",
url = "${FIRSTLEG_SERVICE_URL:${baiyue.clients.firstleg.url:}}")
public interface FirstlegEventClient {
@PostMapping("/internal/firstleg/v1/container-events")
ApiResponse<DeliveryAck> receive(@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody EventEnvelope envelope);
}
lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java
+3 -2
View File
@@ -3,6 +3,7 @@ package com.baiyue.lastleg.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.api.PageResult;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.security.AuthenticatedUser;
import com.baiyue.lastleg.dto.LastlegOrderDetailResponse;
import com.baiyue.lastleg.dto.LastlegOrderRequests;
@@ -60,11 +61,11 @@ public class LastlegOrderController {
@PostMapping("/internal/lastleg/events")
@Operation(summary = "Receive a first-leg loading or actual shipping-node event",
description = "Accepts loading-completed V1/V2 and FirstLegContainerShippingNodeRecorded V1 through the internal token boundary; only monotonic actual PICKUP_READY-or-later facts open overseas PDA receiving.")
public ApiResponse<Integer> receive(
public ApiResponse<DeliveryAck> receive(
@RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
@RequestBody EventEnvelope event) {
tokenGuard.verify(token);
int consumed;
DeliveryAck consumed;
if (EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED.equals(event.eventType())) {
consumed = shippingNodeConsumer.consume(event);
} else if (EventTypes.WALLET_COMMAND_COMPLETED.equals(event.eventType())) {
lastleg-service/src/main/java/com/baiyue/lastleg/financefee/repository/OverseasFeeOutboxRepository.java
+10 -1
View File
@@ -27,8 +27,11 @@ public class OverseasFeeOutboxRepository {
public boolean claim(long id, String worker, LocalDateTime now, LocalDateTime stale) {
return mapper.update(null, new LambdaUpdateWrapper<LastlegEventOutbox>()
.eq(LastlegEventOutbox::getId, id).eq(LastlegEventOutbox::getEventType, EventTypes.BUSINESS_FEE_SOURCE_CHANGED)
.and(status -> status.eq(LastlegEventOutbox::getStatus, "PENDING").or(processing ->
.and(status -> status.eq(LastlegEventOutbox::getStatus, "PENDING")
.and(retry -> retry.isNull(LastlegEventOutbox::getNextRetryAt).or().le(LastlegEventOutbox::getNextRetryAt, now))
.or(processing ->
processing.eq(LastlegEventOutbox::getStatus, "PROCESSING").le(LastlegEventOutbox::getLockedAt, stale)))
.setSql("retry_count = retry_count + CASE WHEN status='PROCESSING' THEN 1 ELSE 0 END")
.set(LastlegEventOutbox::getStatus, "PROCESSING").set(LastlegEventOutbox::getLockedBy, worker)
.set(LastlegEventOutbox::getLockedAt, now)) == 1;
}
@@ -40,6 +43,12 @@ public class OverseasFeeOutboxRepository {
mapper.update(null, claimed(id, worker, "PUBLISHED").set(LastlegEventOutbox::getPublishedAt, now)
.set(LastlegEventOutbox::getNextRetryAt, null).set(LastlegEventOutbox::getLastError, null));
}
public void dead(long id, String worker, int failures, String error) {
mapper.update(null, claimed(id, worker, "DEAD")
.set(LastlegEventOutbox::getRetryCount, failures)
.set(LastlegEventOutbox::getNextRetryAt, null)
.set(LastlegEventOutbox::getLastError, truncate(error)));
}
public void retry(long id, String worker, int retryCount, LocalDateTime next, String error) {
mapper.update(null, claimed(id, worker, "PENDING").set(LastlegEventOutbox::getRetryCount, retryCount)
.set(LastlegEventOutbox::getNextRetryAt, next).set(LastlegEventOutbox::getLastError, truncate(error)));
lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java
+28 -11
View File
@@ -1,7 +1,8 @@
package com.baiyue.lastleg.financefee.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.lastleg.client.EventBridgeClient;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.lastleg.client.DirectEventClient;
import com.baiyue.lastleg.entity.LastlegEventOutbox;
import com.baiyue.lastleg.financefee.repository.OverseasFeeOutboxRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -10,7 +11,6 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.lang.management.ManagementFactory;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.UUID;
@@ -18,28 +18,45 @@ import java.util.UUID;
@Component
public class OverseasFeeOutboxRelay {
@Resource private OverseasFeeOutboxRepository repository;
@Resource private EventBridgeClient client;
@Resource private DirectEventClient client;
@Resource private ObjectMapper objectMapper;
@Value("${baiyue.security.internal-service-token:}") private String token;
private final String worker = ManagementFactory.getRuntimeMXBean().getName() + ':' + UUID.randomUUID();
@Value("${baiyue.event-delivery.max-attempts:12}") private int maxAttempts = 12;
@Value("${baiyue.event-delivery.base-delay-seconds:5}") private long baseDelaySeconds = 5;
@Value("${baiyue.event-delivery.max-delay-seconds:1800}") private long maxDelaySeconds = 1800;
@Value("${baiyue.event-delivery.jitter:0.1}") private double retryJitter = 0.1;
@Value("${baiyue.event-delivery.lease-seconds:60}") private long leaseSeconds = 60;
@Scheduled(fixedDelayString = "${baiyue.finance-fee-relay.fixed-delay:3000}")
public void relay() {
if (leaseSeconds < 1) throw new IllegalArgumentException("Lease duration must be positive");
var policy = new BoundedRetryPolicy(maxAttempts, baseDelaySeconds, maxDelaySeconds, retryJitter);
LocalDateTime now = LocalDateTime.now(ZoneOffset.UTC);
for (LastlegEventOutbox candidate : repository.pending(50, now, now.minusSeconds(30))) {
if (!repository.claim(candidate.getId(), worker, now, now.minusSeconds(30))) continue;
for (LastlegEventOutbox candidate : repository.pending(50, now, now.minusSeconds(leaseSeconds))) {
String worker = UUID.randomUUID().toString();
now = LocalDateTime.now(ZoneOffset.UTC);
if (!repository.claim(candidate.getId(), worker, now, now.minusSeconds(leaseSeconds))) continue;
LastlegEventOutbox row = repository.claimed(candidate.getId(), worker);
if (row == null) continue;
int failures = row.getRetryCount() == null ? 0 : row.getRetryCount();
if (policy.exhausted(failures)) {
repository.dead(row.getId(), worker, failures, "自动投递次数已耗尽（含租约超时）");
continue;
}
try {
EventEnvelope envelope = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class).requireValid();
var response = client.receive(token, envelope);
if (response == null || !"200".equals(response.code()) || response.data() == null
|| !envelope.eventId().equals(response.data().eventId()))
throw new IllegalStateException("Event Bridge未确认海外费用事件");
if (response == null || !response.successful() || response.data() == null)
throw new IllegalStateException("Finance未确认费用事件");
response.data().requireMatches(envelope);
repository.published(row.getId(), worker, LocalDateTime.now(ZoneOffset.UTC));
} catch (Exception exception) {
int retries = row.getRetryCount() == null ? 1 : row.getRetryCount() + 1;
repository.retry(row.getId(), worker, retries, now.plusSeconds(Math.min(900, 5L << Math.min(retries, 7))), exception.getMessage());
int retries = failures + 1;
if (policy.exhausted(retries)) repository.dead(row.getId(), worker, retries, exception.getMessage());
else repository.retry(row.getId(), worker, retries, LocalDateTime.now(ZoneOffset.UTC).plus(policy.delayAfterFailure(retries)), exception.getMessage());
}
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java
+30 -16
View File
@@ -1,8 +1,10 @@
package com.baiyue.lastleg.inboundplan.pda.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelopeValidator;
import com.baiyue.lastleg.client.EventBridgeClient;
import com.baiyue.lastleg.client.DirectEventClient;
import com.baiyue.lastleg.entity.LastlegEventOutbox;
import com.baiyue.lastleg.repository.UnloadCompletedOutboxRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -15,26 +17,40 @@ import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.UUID;
/** 至少一次投递海外到国内业务事件；国内事件桥按 eventId 幂等接收。 */
/** 至少一次投递海外到国内业务事件；消费服务按 eventId 幂等接收。 */
@Component
@Slf4j
public class InboundUnloadCompletedOutboxRelay {
@Resource private UnloadCompletedOutboxRepository repository;
@Resource private EventBridgeClient client;
@Resource private DirectEventClient client;
@Resource private ObjectMapper objectMapper;
@Value("${baiyue.security.internal-service-token}") private String internalToken;
@Value("${baiyue.firstleg.outbox-relay.lease-seconds:60}") private long leaseSeconds;
@Value("${baiyue.firstleg.outbox-relay.retry-seconds:30}") private long retrySeconds;
private final String workerId = UUID.randomUUID().toString();
@Value("${baiyue.event-delivery.lease-seconds:60}") private long leaseSeconds = 60;
@Value("${baiyue.event-delivery.max-attempts:12}") private int maxAttempts = 12;
@Value("${baiyue.event-delivery.base-delay-seconds:5}") private long baseDelaySeconds = 5;
@Value("${baiyue.event-delivery.max-delay-seconds:1800}") private long maxDelaySeconds = 1800;
@Value("${baiyue.event-delivery.jitter:0.1}") private double retryJitter = 0.1;
@Scheduled(fixedDelayString = "${baiyue.firstleg.outbox-relay.fixed-delay-ms:5000}")
public void relay() {
if (leaseSeconds < 1) throw new IllegalArgumentException("Lease duration must be positive");
var policy = new BoundedRetryPolicy(maxAttempts, baseDelaySeconds, maxDelaySeconds, retryJitter);
LocalDateTime now = LocalDateTime.now();
LocalDateTime expiredLeaseBefore = now.minusSeconds(Math.max(leaseSeconds, 10));
LocalDateTime expiredLeaseBefore = now.minusSeconds(leaseSeconds);
for (LastlegEventOutbox candidate : repository.pending(50, now, expiredLeaseBefore)) {
String workerId = UUID.randomUUID().toString();
now = LocalDateTime.now();
expiredLeaseBefore = now.minusSeconds(leaseSeconds);
if (!repository.claim(candidate.getId(), workerId, now, expiredLeaseBefore)) continue;
LastlegEventOutbox row = repository.claimed(candidate.getId(), workerId);
if (row == null) continue;
int failures = row.getRetryCount() == null ? 0 : row.getRetryCount();
if (policy.exhausted(failures)) {
repository.dead(row.getId(), workerId, failures, "自动投递次数已耗尽（含租约超时）");
continue;
}
try {
EventEnvelope event = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class);
EventEnvelopeValidator.requireValid(event);
@@ -42,7 +58,8 @@ public class InboundUnloadCompletedOutboxRelay {
requireAcknowledgement(event, response);
repository.published(row.getId(), workerId, LocalDateTime.now());
} catch (Exception exception) {
repository.retry(row.getId(), workerId, LocalDateTime.now().plusSeconds(Math.max(retrySeconds, 1)),
if (policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, exception.getMessage());
else repository.retry(row.getId(), workerId, LocalDateTime.now().plus(policy.delayAfterFailure(failures + 1)),
exception.getMessage());
log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), exception.getMessage());
}
@@ -50,16 +67,13 @@ public class InboundUnloadCompletedOutboxRelay {
}
private void requireAcknowledgement(EventEnvelope event,
com.baiyue.common.core.api.ApiResponse<EventBridgeClient.Acknowledgement> response) {
if (response == null) throw new IllegalStateException("国内事件桥返回空响应");
com.baiyue.common.core.api.ApiResponse<DeliveryAck> response) {
if (response == null) throw new IllegalStateException("消费服务返回空响应");
if (!response.successful()) {
throw new IllegalStateException("国内事件桥拒绝海外事件: code=" + response.code()
throw new IllegalStateException("消费服务拒绝海外事件: code=" + response.code()
+ ", message=" + response.message() + ", requestId=" + response.requestId());
}
if (response.data() == null) throw new IllegalStateException("国内事件桥成功响应缺少确认数据");
if (!event.eventId().equals(response.data().eventId())) {
throw new IllegalStateException("国内事件桥确认eventId不一致: expected=" + event.eventId()
+ ", actual=" + response.data().eventId());
}
if (response.data() == null) throw new IllegalStateException("消费服务成功响应缺少确认数据");
response.data().requireMatches(event);
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/repository/UnloadCompletedOutboxRepository.java
+9
View File
@@ -41,6 +41,7 @@ public class UnloadCompletedOutboxRepository {
.or().le(LastlegEventOutbox::getNextRetryAt, now))
.or(stale -> stale.eq(LastlegEventOutbox::getStatus, "PROCESSING")
.le(LastlegEventOutbox::getLockedAt, expiredLeaseBefore)))
.setSql("retry_count = retry_count + CASE WHEN status='PROCESSING' THEN 1 ELSE 0 END")
.set(LastlegEventOutbox::getStatus, "PROCESSING")
.set(LastlegEventOutbox::getLockedBy, worker)
.set(LastlegEventOutbox::getLockedAt, now)) == 1;
@@ -64,6 +65,14 @@ public class UnloadCompletedOutboxRepository {
.set(LastlegEventOutbox::getLastError, null));
}
public void dead(long id, String worker, int failures, String error) {
mapper.update(null, claimedBase(id, worker)
.set(LastlegEventOutbox::getStatus, "DEAD")
.set(LastlegEventOutbox::getRetryCount, failures)
.set(LastlegEventOutbox::getNextRetryAt, null)
.set(LastlegEventOutbox::getLockedBy, null).set(LastlegEventOutbox::getLockedAt, null)
.set(LastlegEventOutbox::getLastError, truncate(error)));
}
public void retry(long id, String worker, LocalDateTime retryAt, String error) {
mapper.update(null, claimedBase(id, worker)
.setSql("retry_count = retry_count + 1")
lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java
+6 -4
View File
@@ -51,7 +51,7 @@ public class LoadingCompletedConsumer {
@Resource private ContainerLoadingEvidenceMapper loadingEvidence;
@Transactional(isolation = Isolation.READ_COMMITTED)
public int consume(EventEnvelope envelope) {
public DeliveryAck consume(EventEnvelope envelope) {
envelope.requireValid();
if (!EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(envelope.eventType())
|| (envelope.eventVersion() != 1 && envelope.eventVersion() != 2 && envelope.eventVersion() != 3))
@@ -72,7 +72,8 @@ public class LoadingCompletedConsumer {
LastlegEventInbox existing=inboxMapper.selectById(envelope.eventId());
if(existing!=null) {
if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new IllegalArgumentException("事件摘要冲突");
return countOrders(envelope,containerNo);
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
LastlegEventInbox inbox=new LastlegEventInbox();
@@ -85,7 +86,8 @@ public class LoadingCompletedConsumer {
|| !raced.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("事件摘要冲突", exception);
}
return countOrders(envelope,containerNo);
if (!"PROCESSED".equals(raced.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
projectLoadingEvidence(envelope, sourceContainerBookingId);
@@ -102,7 +104,7 @@ public class LoadingCompletedConsumer {
sourceContainerBookingId, owner);
}
inbox.setStatus("PROCESSED"); inbox.setProcessedAt(LocalDateTime.now()); inboxMapper.updateById(inbox);
return payload.orders().size();
return DeliveryAck.of(envelope, DeliveryAck.Outcome.APPLIED);
}
private void projectLoadingEvidence(EventEnvelope envelope, String sourceContainerBookingId) {
lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java
+10 -6
View File
@@ -1,6 +1,7 @@
package com.baiyue.lastleg.shipping.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.common.event.EventTypes;
@@ -28,7 +29,7 @@ public class ContainerShippingNodeConsumer {
@Resource private ObjectMapper objectMapper;
@Transactional(isolation = Isolation.READ_COMMITTED)
public int consume(EventEnvelope envelope) {
public DeliveryAck consume(EventEnvelope envelope) {
validate(envelope);
FirstLegContainerShippingNodeRecordedV1 payload = payload(envelope);
validateEnvelopePayload(envelope, payload);
@@ -37,7 +38,8 @@ public class ContainerShippingNodeConsumer {
if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("船节点事件摘要冲突");
}
return 1;
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
LastlegEventInbox inbox = inbox(envelope);
try { inboxes.insert(inbox); }
@@ -46,7 +48,8 @@ public class ContainerShippingNodeConsumer {
if (concurrent == null || !concurrent.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("船节点事件摘要冲突", raced);
}
return 1;
if (!"PROCESSED".equals(concurrent.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
gates.initialize(payload.containerBookingId(), payload.oceanBookingId(),
@@ -55,13 +58,13 @@ public class ContainerShippingNodeConsumer {
if (gate == null || !payload.containerNo().equals(gate.getContainerNo())) {
throw new IllegalArgumentException("船节点柜号快照与装柜交接不一致");
}
apply(gate, envelope, payload);
boolean applied = apply(gate, envelope, payload);
gates.updateById(gate);
inbox.setStatus("PROCESSED"); inbox.setProcessedAt(LocalDateTime.now()); inboxes.updateById(inbox);
return 1;
return DeliveryAck.of(envelope, applied ? DeliveryAck.Outcome.APPLIED : DeliveryAck.Outcome.STALE);
}
private void apply(ContainerShippingGate gate, EventEnvelope envelope,
private boolean apply(ContainerShippingGate gate, EventEnvelope envelope,
FirstLegContainerShippingNodeRecordedV1 payload) {
long currentVersion = gate.getAggregateVersion() == null ? 0 : gate.getAggregateVersion();
int currentSequence = gate.getNodeSequence() == null ? 0 : gate.getNodeSequence();
@@ -88,6 +91,7 @@ public class ContainerShippingNodeConsumer {
gate.setHandoverVersion(Math.max(gate.getHandoverVersion() == null ? 0 : gate.getHandoverVersion(),
envelope.handoverVersion()));
gate.setUpdatedAt(LocalDateTime.now());
return accepted;
}
private void validate(EventEnvelope envelope) {
lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java
+11 -6
View File
@@ -1,6 +1,7 @@
package com.baiyue.lastleg.wallet;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelopeValidator;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
@@ -26,23 +27,27 @@ public class WalletCommandCompletedConsumer {
private final ObjectMapper objectMapper;
private final List<WalletResultHandler> handlers;
@Transactional
public int consume(EventEnvelope envelope) {
@Transactional(isolation = org.springframework.transaction.annotation.Isolation.READ_COMMITTED)
public DeliveryAck consume(EventEnvelope envelope) {
WalletCommandCompletedV1 result = validate(envelope);
List<String> existing = jdbc.queryForList("SELECT payload_hash FROM ll_wallet_event_inbox WHERE event_id=?", String.class, envelope.eventId());
List<String> existing = jdbc.queryForList("SELECT payload_hash FROM ll_wallet_event_inbox WHERE event_id=? AND status='PROCESSED'", String.class, envelope.eventId());
if (!existing.isEmpty()) {
if (!existing.get(0).equals(envelope.payloadHash())) throw bad("同一资金结果事件ID对应不同载荷");
return 0;
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
try {
jdbc.update("INSERT INTO ll_wallet_event_inbox(event_id,command_event_id,payload_hash,payload_json,status,received_at) VALUES(?,?,?,CAST(? AS JSON),'RECEIVED',NOW(3))",
envelope.eventId(), result.commandEventId(), envelope.payloadHash(), json(envelope));
} catch (DuplicateKeyException exception) { return 0; }
} catch (DuplicateKeyException exception) {
List<String> raced = jdbc.queryForList("SELECT payload_hash FROM ll_wallet_event_inbox WHERE event_id=? AND status='PROCESSED'", String.class, envelope.eventId());
if (raced.size() != 1 || !raced.get(0).equals(envelope.payloadHash())) throw bad("资金结果事件身份冲突或尚未完成");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
WalletResultHandler handler = handlers.stream().filter(value -> value.supports(result.businessType())).findFirst()
.orElseThrow(() -> bad("没有可处理该资金结果的业务模块"));
handler.handle(result);
jdbc.update("UPDATE ll_wallet_event_inbox SET status='PROCESSED',processed_at=NOW(3) WHERE event_id=?", envelope.eventId());
return 1;
return DeliveryAck.of(envelope, DeliveryAck.Outcome.APPLIED);
}
private WalletCommandCompletedV1 validate(EventEnvelope envelope) {