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-25 19:07:34 +08:00
parent 65c01f8b3d
commit 7d36759f3f
28 changed files with 613 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/HttpFirstlegEventClient.java
+5 -2
View File
@@ -4,6 +4,7 @@ import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.RegionCode;
import com.baiyue.eventbridge.config.EventDispatcherProperties;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.trace.EventBridgeHttpTracePropagation;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PostConstruct;
@@ -20,6 +21,7 @@ import java.net.http.HttpResponse;
public class HttpFirstlegEventClient implements FirstlegEventClient {
@Resource private ObjectMapper objectMapper;
@Resource private EventDispatcherProperties properties;
@Resource private EventBridgeHttpTracePropagation tracePropagation;
@Value("${baiyue.event-dispatcher.firstleg-consumer-url:http://127.0.0.1:9203/internal/firstleg/v1/container-events}")
private URI consumerUrl;
@Value("${baiyue.security.internal-service-token:}") private String serviceToken;
@@ -37,9 +39,10 @@ public class HttpFirstlegEventClient implements FirstlegEventClient {
row.getCustomerId(), row.getAggregateType(), row.getAggregateId(), row.getAggregateVersion(),
row.getHandoverVersion(), row.getTraceId(), row.getCorrelationId(), row.getCausationId(),
objectMapper.readTree(row.getPayloadJson()), row.getPayloadHash());
HttpRequest request = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
.header("Content-Type", "application/json")
.header("X-Internal-Service-Token", serviceToken)
.header("X-Internal-Service-Token", serviceToken);
HttpRequest request = tracePropagation.apply(requestBuilder)
.POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(envelope))).build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
JsonNode responseBody;
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/HttpLastlegEventClient.java
+5 -2
View File
@@ -4,6 +4,7 @@ import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.RegionCode;
import com.baiyue.eventbridge.config.EventDispatcherProperties;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.trace.EventBridgeHttpTracePropagation;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Value;
@@ -19,6 +20,7 @@ import java.net.http.HttpResponse;
public class HttpLastlegEventClient implements LastlegEventClient {
@Resource private ObjectMapper objectMapper;
@Resource private EventDispatcherProperties properties;
@Resource private EventBridgeHttpTracePropagation tracePropagation;
@Value("${baiyue.security.internal-service-token:}")
private String serviceToken;
@@ -38,10 +40,11 @@ public class HttpLastlegEventClient implements LastlegEventClient {
row.getHandoverVersion(), row.getTraceId(), row.getCorrelationId(), row.getCausationId(),
objectMapper.readTree(row.getPayloadJson()), row.getPayloadHash());
String body = objectMapper.writeValueAsString(envelope);
HttpRequest request = HttpRequest.newBuilder(properties.consumerUrl())
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(properties.consumerUrl())
.timeout(properties.requestTimeout())
.header("Content-Type", "application/json")
.header("X-Internal-Service-Token", serviceToken)
.header("X-Internal-Service-Token", serviceToken);
HttpRequest request = tracePropagation.apply(requestBuilder)
.POST(HttpRequest.BodyPublishers.ofString(body))
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/HttpSecurityProjectionEventClient.java
+6 -2
View File
@@ -4,9 +4,11 @@ import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.RegionCode;
import com.baiyue.eventbridge.config.EventDispatcherProperties;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.trace.EventBridgeHttpTracePropagation;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@@ -19,6 +21,7 @@ import java.net.http.HttpResponse;
public class HttpSecurityProjectionEventClient implements SecurityProjectionEventClient {
private final ObjectMapper objectMapper;
private final EventDispatcherProperties properties;
@Resource private EventBridgeHttpTracePropagation tracePropagation;
@Value("${baiyue.event-dispatcher.security-consumer-url}") private URI consumerUrl;
@Value("${baiyue.security.internal-service-token:}") private String serviceToken;
private HttpClient httpClient;
@@ -39,9 +42,10 @@ public class HttpSecurityProjectionEventClient implements SecurityProjectionEven
row.getCustomerId(), row.getAggregateType(), row.getAggregateId(), row.getAggregateVersion(),
row.getHandoverVersion(), row.getTraceId(), row.getCorrelationId(), row.getCausationId(),
objectMapper.readTree(row.getPayloadJson()), row.getPayloadHash());
HttpRequest request = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
.header("Content-Type", "application/json")
.header("X-Internal-Service-Token", serviceToken)
.header("X-Internal-Service-Token", serviceToken);
HttpRequest request = tracePropagation.apply(requestBuilder)
.POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(envelope))).build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
JsonNode body;
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/OverseasEventDispatcher.java
+13 -3
View File
@@ -7,10 +7,13 @@ import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.mapper.BridgeDeliveryAttemptMapper;
import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
import com.baiyue.eventbridge.metrics.EventBridgeMetrics;
import com.baiyue.eventbridge.trace.EventBridgeTraceScope;
import com.baiyue.common.web.CorrelationContext;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.slf4j.MDC;
import java.lang.management.ManagementFactory;
import java.time.LocalDateTime;
@@ -42,6 +45,12 @@ public class OverseasEventDispatcher {
}
BridgeInbox event = inboxMapper.selectById(eventId);
try (EventBridgeTraceScope ignored = EventBridgeTraceScope.open(event)) {
dispatchClaimed(eventId, event);
}
}
private void dispatchClaimed(String eventId, BridgeInbox event) {
int retryAttempt = event.getRetryCount() == null ? 1 : event.getRetryCount() + 1;
int deliveryAttemptNo = inboxMapper.nextAttemptNo(eventId);
BridgeDeliveryAttempt attempt = startAttempt(event, deliveryAttemptNo);
@@ -57,8 +66,8 @@ public class OverseasEventDispatcher {
: "FAILED_LEASE_LOST";
finishAttempt(attempt, attemptStatus, null, error);
metrics.recordTransport(event, EventBridgeMetrics.TransportResult.valueOf(attemptStatus));
log.warn("跨区事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, reason={}",
eventId, retryAttempt, deliveryAttemptNo, error);
log.warn("跨区事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, traceId={}, reason={}",
eventId, retryAttempt, deliveryAttemptNo, MDC.get(CorrelationContext.TRACE_ID), error);
return;
}
@@ -66,7 +75,8 @@ public class OverseasEventDispatcher {
String error = "下游已确认成功，但事件租约已失效；保留当前 Inbox 状态并由新租约持有者幂等续投";
finishAttempt(attempt, "ACKNOWLEDGED_LEASE_LOST", responseCode, error);
metrics.recordTransport(event, EventBridgeMetrics.TransportResult.ACKNOWLEDGED_LEASE_LOST);
log.warn("跨区事件成功确认时租约已失效 eventId={}, deliveryAttempt={}", eventId, deliveryAttemptNo);
log.warn("跨区事件成功确认时租约已失效 eventId={}, deliveryAttempt={}, traceId={}",
eventId, deliveryAttemptNo, MDC.get(CorrelationContext.TRACE_ID));
return;
}
finishAttempt(attempt, "DELIVERED", responseCode, null);
event-bridge-service/src/main/java/com/baiyue/eventbridge/trace/EventBridgeHttpTracePropagation.java
+19
View File
@@ -0,0 +1,19 @@
package com.baiyue.eventbridge.trace;
import com.baiyue.common.web.CorrelationContext;
import com.baiyue.common.web.OutboundTraceHeaders;
import org.springframework.stereotype.Component;
import java.net.http.HttpRequest;
/** Applies a fresh child span to one raw JDK HttpClient request. */
@Component
public class EventBridgeHttpTracePropagation {
public HttpRequest.Builder apply(HttpRequest.Builder builder) {
OutboundTraceHeaders outbound = OutboundTraceHeaders.fromCurrentOrNew();
return builder
.setHeader(CorrelationContext.REQUEST_ID_HEADER, outbound.requestId())
.setHeader(CorrelationContext.TRACE_ID_HEADER, outbound.traceId())
.setHeader(CorrelationContext.TRACE_PARENT_HEADER, outbound.traceParent());
}
}
event-bridge-service/src/main/java/com/baiyue/eventbridge/trace/EventBridgeTraceScope.java
+35
View File
@@ -0,0 +1,35 @@
package com.baiyue.eventbridge.trace;
import com.baiyue.common.web.CorrelationContext;
import com.baiyue.common.web.OutboundTraceHeaders;
import com.baiyue.eventbridge.domain.BridgeInbox;
import org.slf4j.MDC;
import java.util.Map;
/** Establishes one trusted MDC context for an Event Bridge delivery attempt. */
public final class EventBridgeTraceScope implements AutoCloseable {
private final Map<String, String> previous;
private EventBridgeTraceScope(Map<String, String> previous) {
this.previous = previous;
}
public static EventBridgeTraceScope open(BridgeInbox event) {
Map<String, String> previous = MDC.getCopyOfContextMap();
OutboundTraceHeaders local = OutboundTraceHeaders.fromTrustedTraceOrNew(event.getTraceId());
MDC.put(CorrelationContext.REQUEST_ID, local.requestId());
MDC.put(CorrelationContext.TRACE_ID, local.traceId());
MDC.put(CorrelationContext.SPAN_ID, local.spanId());
return new EventBridgeTraceScope(previous);
}
@Override
public void close() {
if (previous == null) {
MDC.clear();
} else {
MDC.setContextMap(previous);
}
}
}