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-20 18:02:38 +08:00
parent b55b049e77
commit 1cd8b6a473
210 changed files with 6974 additions and 825 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event-bridge-service/src/main/java/com/baiyue/eventbridge/controller/EventBridgeController.java
+1 -1
View File
@@ -36,7 +36,7 @@ public class EventBridgeController {
@Value("${baiyue.region:OVERSEAS}")
private String region;
@Operation(summary = "幂等接收事件", description = "当前仅完成数据库接收骨架，不代表生产跨区传输和 mTLS 已开通")
@Operation(summary = "幂等接收事件", description = "按当前部署区域校验目标区域、事件信封与 payloadHash，幂等持久化后进入租约保护的重试投递队列")
@ApiResponses({
@io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "200", description = "事件已接收或判定为重复"),
@io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "401", description = "内部服务令牌无效"),
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java
+23 -5
View File
@@ -14,6 +14,7 @@ import org.springframework.transaction.annotation.Transactional;
import jakarta.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Objects;
/**
* 入站事件校验与幂等落库服务。
@@ -32,9 +33,8 @@ public class EventInboxService {
if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
}
boolean currentHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash());
boolean legacyHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalizeLegacy(envelope.payload()), envelope.payloadHash());
if (!currentHash && !legacyHash) {
if (!EventPayloadHash.matches(
EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
}
BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());
@@ -55,12 +55,30 @@ public class EventInboxService {
}
private EventReceiveResult duplicateOrConflict(BridgeInbox existing, EventEnvelope envelope) {
if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("相同 eventId 对应的 payloadHash 不一致，拒绝覆盖既有事件");
if (!sameEnvelope(existing, envelope)) {
throw new IllegalArgumentException("相同 eventId 对应的事件信封或 payloadHash 不一致，拒绝覆盖既有事件");
}
return new EventReceiveResult(envelope.eventId(), "重复", existing.getReceivedAt());
}
private boolean sameEnvelope(BridgeInbox existing, EventEnvelope envelope) {
return existing.getPayloadHash() != null
&& existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())
&& Objects.equals(existing.getEventType(), envelope.eventType())
&& Objects.equals(existing.getEventVersion(), envelope.eventVersion())
&& Objects.equals(existing.getSourceRegion(), envelope.sourceRegion().name())
&& Objects.equals(existing.getTargetRegion(), envelope.targetRegion().name())
&& Objects.equals(existing.getCustomerId(), envelope.tenantId())
&& Objects.equals(existing.getAggregateType(), envelope.aggregateType())
&& Objects.equals(existing.getAggregateId(), envelope.aggregateId())
&& Objects.equals(existing.getAggregateVersion(), envelope.aggregateVersion())
&& Objects.equals(existing.getHandoverVersion(), envelope.handoverVersion())
&& Objects.equals(existing.getOccurredAt(), envelope.occurredAt())
&& Objects.equals(existing.getTraceId(), envelope.traceId())
&& Objects.equals(existing.getCorrelationId(), envelope.correlationId())
&& Objects.equals(existing.getCausationId(), envelope.causationId());
}
private BridgeInbox toInbox(EventEnvelope envelope) {
BridgeInbox inbox = new BridgeInbox();
inbox.setEventId(envelope.eventId());
event-bridge-service/src/main/resources/application.yml
+2 -2
View File
@@ -19,7 +19,7 @@ spring:
password: ${EVENT_BRIDGE_DB_PASSWORD:123456}
cloud:
nacos:
server-addr: ${NACOS_SERVER_ADDR:127.0.0.1:8858}
server-addr: ${NACOS_SERVER_ADDR:127.0.0.1:8848}
username: ${NACOS_USERNAME:nacos}
password: ${NACOS_PASSWORD:nacos}
discovery:
@@ -31,7 +31,7 @@ spring:
baiyue:
region: ${REGION_CODE:OVERSEAS}
event-dispatcher:
consumer-url: ${LASTLEG_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/events}
consumer-url: ${EVENT_CONSUMER_URL:${LASTLEG_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/events}}
fixed-delay: ${EVENT_DISPATCH_FIXED_DELAY:3000}
batch-size: ${EVENT_DISPATCH_BATCH_SIZE:50}
max-attempts: ${EVENT_DISPATCH_MAX_ATTEMPTS:10}
event-bridge-service/src/test/java/com/baiyue/eventbridge/service/EventInboxServiceTest.java
+36 -7
View File
@@ -36,7 +36,7 @@ class EventInboxServiceTest {
}
@Test
void 数字规范化上线前的历史摘要仍可接收() {
void 拒绝旧数字规范化摘要() {
EventEnvelope source = envelope("sha256:placeholder");
source.payload().withObject("/snapshot").put("weight", 30.0).put("volume", 0.3150);
String legacyHash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalizeLegacy(source.payload()));
@@ -45,13 +45,15 @@ class EventInboxServiceTest {
source.aggregateType(), source.aggregateId(), source.aggregateVersion(), source.handoverVersion(),
source.traceId(), source.correlationId(), source.causationId(), source.payload(), legacyHash);
assertThat(service.receive(envelope).result()).isEqualTo("已接收");
assertThatThrownBy(() -> service.receive(envelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("payloadHash");
}
@Test
void 相同事件和摘要重复接收不再次写入() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
BridgeInbox existing = existing(envelope.payloadHash());
BridgeInbox existing = existing(envelope, envelope.payloadHash());
when(store.findByEventId(envelope.eventId())).thenReturn(existing);
assertThat(service.receive(envelope).result()).isEqualTo("重复");
@@ -60,17 +62,31 @@ class EventInboxServiceTest {
@Test
void 相同事件但摘要不同时拒绝覆盖() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
when(store.findByEventId(envelope.eventId())).thenReturn(existing("sha256:" + "0".repeat(64)));
when(store.findByEventId(envelope.eventId())).thenReturn(existing(envelope, "sha256:" + "0".repeat(64)));
assertThatThrownBy(() -> service.receive(envelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("拒绝覆盖");
}
@Test
void 相同事件和摘要但聚合标识不同时拒绝覆盖() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
EventEnvelope conflicting = new EventEnvelope(envelope.eventId(), envelope.eventType(), envelope.eventVersion(),
envelope.occurredAt(), envelope.sourceRegion(), envelope.targetRegion(), envelope.tenantId(),
envelope.aggregateType(), "DIFFERENT", envelope.aggregateVersion(), envelope.handoverVersion(),
envelope.traceId(), envelope.correlationId(), envelope.causationId(), envelope.payload(), envelope.payloadHash());
when(store.findByEventId(conflicting.eventId())).thenReturn(existing(envelope, envelope.payloadHash()));
assertThatThrownBy(() -> service.receive(conflicting))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("事件信封");
}
@Test
void 并发插入冲突后按既有记录判定为重复() {
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
when(store.findByEventId(envelope.eventId())).thenReturn(null, existing(envelope.payloadHash()));
when(store.findByEventId(envelope.eventId())).thenReturn(null, existing(envelope, envelope.payloadHash()));
doThrow(new DuplicateKeyException("并发重复")).when(store).insert(any());
assertThat(service.receive(envelope).result()).isEqualTo("重复");
@@ -109,9 +125,22 @@ class EventInboxServiceTest {
source.correlationId(), source.causationId(), source.payload(), hash);
}
private BridgeInbox existing(String hash) {
private BridgeInbox existing(EventEnvelope envelope, String hash) {
BridgeInbox inbox = new BridgeInbox();
inbox.setEventId("550e8400-e29b-41d4-a716-446655440000");
inbox.setEventId(envelope.eventId());
inbox.setEventType(envelope.eventType());
inbox.setEventVersion(envelope.eventVersion());
inbox.setSourceRegion(envelope.sourceRegion().name());
inbox.setTargetRegion(envelope.targetRegion().name());
inbox.setCustomerId(envelope.tenantId());
inbox.setAggregateType(envelope.aggregateType());
inbox.setAggregateId(envelope.aggregateId());
inbox.setAggregateVersion(envelope.aggregateVersion());
inbox.setHandoverVersion(envelope.handoverVersion());
inbox.setOccurredAt(envelope.occurredAt());
inbox.setTraceId(envelope.traceId());
inbox.setCorrelationId(envelope.correlationId());
inbox.setCausationId(envelope.causationId());
inbox.setPayloadHash(hash);
inbox.setReceivedAt(LocalDateTime.of(2026, 7, 29, 18, 0));
return inbox;