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

chore(lastleg): instrument appointment latency from restored 662dead baseline

Browse Source
This commit is contained in:
ouyangxq
2026-10-18 11:43:10 +08:00
parent b0acbfbb34
commit 5cd108b39a
35 changed files with 951 additions and 780 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentplanning/AppointmentCargoReadRepository.java
-16
View File
@@ -1,16 +0,0 @@
package com.baiyue.lastleg.appointmentplanning;
import org.springframework.stereotype.Repository;
import java.util.List;
/** 单次生成只批量读取核实与分配原料，业务解析仍按实际访问顺序。 */
@Repository
public class AppointmentCargoReadRepository {
private final AppointmentCargoVerificationMapper mapper;
/** 复用本域既有批量查询。 */
public AppointmentCargoReadRepository(AppointmentCargoVerificationMapper mapper) { this.mapper=mapper; }
/** 取得每客户来源的最新核实记录，暂不解析货件 JSON。 */
public List<AppointmentCargoVerification> latest(List<AppointmentCoverageService.SourceRef> refs) { return refs.isEmpty()?List.of():mapper.latestBatch(refs); }
/** 分配数量仅取活动明细，未知数量保持单独计数。 */
public List<AppointmentCargoVerificationMapper.SourceAllocation> allocations(List<AppointmentCoverageService.SourceRef> refs) { return refs.isEmpty()?List.of():mapper.allocatedBatch(refs); }
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentplanning/AppointmentCargoVerificationService.java
-18
View File
@@ -21,7 +21,6 @@ public class AppointmentCargoVerificationService {
private final AppointmentSourceService sources;
private final AppointmentCargoVerificationMapper mapper;
private final ObjectMapper json;
@jakarta.annotation.Resource private AppointmentCargoReadRepository readRepository;
public AppointmentCargoVerificationService(AppointmentSourceService sources, AppointmentCargoVerificationMapper mapper,
ObjectMapper json) {
@@ -51,23 +50,6 @@ public class AppointmentCargoVerificationService {
public record Inspection(AppointmentCoverageService.SourceRef reference,Context context,ResponseStatusException error) {
public Context requireContext() { if(error!=null) throw error; return context; }
}
/** 只预取本次已授权来源的原料；核实 JSON 与数量判断按原 anyMatch 实际访问顺序执行。 */
public java.util.function.Function<AppointmentSourceService.Source,Context> prepareInspection(List<AppointmentSourceService.Source> resolved) {
var refs=resolved.stream().map(s->new AppointmentCoverageService.SourceRef(s.customerId(),s.sourceKey(),s.destinationKey())).distinct().toList();
var latest=new HashMap<List<String>,AppointmentCargoVerification>();
var allocations=new HashMap<List<String>,AppointmentCargoVerificationMapper.Allocation>();
for(int start=0;start<refs.size();start+=100) {
var batch=refs.subList(start,Math.min(start+100,refs.size()));
readRepository.latest(batch).forEach(r->latest.put(List.of(r.getCustomerId(),r.getSourceKey()),r));
readRepository.allocations(batch).forEach(r->allocations.put(List.of(r.customerId(),r.sourceKey()),new AppointmentCargoVerificationMapper.Allocation(r.packages(),r.unknownItems())));
}
var allowed=new HashSet<>(resolved);
return source->{
if(!allowed.contains(source)) throw new IllegalArgumentException("货物核实来源不属于本次读取范围");
var key=List.of(source.customerId(),source.sourceKey());
return inspect(source,latest.get(key),allocations.getOrDefault(key,new AppointmentCargoVerificationMapper.Allocation(0,0)));
};
}
/** Shares source and warehouse reads within this call only; never caches mutable business facts across requests. */
public List<Inspection> inspectBatch(List<AppointmentCoverageService.SourceRef> refs) {
var result=new ArrayList<Inspection>();
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentplanning/AppointmentSourceReadRepository.java
-26
View File
@@ -1,26 +0,0 @@
package com.baiyue.lastleg.appointmentplanning;
import com.baiyue.lastleg.entity.LastlegOrder;
import com.baiyue.lastleg.mapper.LastlegOrderMapper;
import com.baiyue.lastleg.shipping.entity.ContainerShippingGate;
import com.baiyue.lastleg.shipping.mapper.ContainerShippingGateMapper;
import org.springframework.stereotype.Repository;
import java.util.List;
/** 同次需求生成的来源原料读取，不保存跨调用状态。 */
@Repository
public class AppointmentSourceReadRepository {
private final LastlegOrderMapper orders;
private final ContainerShippingGateMapper shipping;
private final AppointmentCoverageMapper destinations;
/** 仅组合本域既有批量持久查询。 */
public AppointmentSourceReadRepository(LastlegOrderMapper orders,ContainerShippingGateMapper shipping,AppointmentCoverageMapper destinations) {
this.orders=orders;this.shipping=shipping;this.destinations=destinations;
}
/** 读取所选订单，客户归属继续由来源服务逐需求核验。 */
public List<LastlegOrder> orders(List<Long> ids) { return ids.isEmpty()?List.of():orders.selectBatchIds(ids); }
/** 读取实际船节点门禁，不从计划时间推断。 */
public List<ContainerShippingGate> gates(List<String> ids) { return ids.isEmpty()?List.of():shipping.selectBatchIds(ids); }
/** 转运目的地仍取原业务权威关系。 */
public List<AppointmentCoverageMapper.Source> destinations(List<AppointmentCoverageMapper.OrderRef> refs) { return refs.isEmpty()?List.of():destinations.sources(refs); }
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentplanning/AppointmentSourceService.java
-85
View File
@@ -26,7 +26,6 @@ public class AppointmentSourceService {
private final AppointmentWarehouseClient warehouses;
private final String token;
@jakarta.annotation.Resource private AppointmentCoverageMapper destinationFacts;
@jakarta.annotation.Resource private AppointmentSourceReadRepository readRepository;
public AppointmentSourceService(LastlegOrderMapper orders, ContainerShippingGateMapper shipping,
AppointmentExternalSourceGateway external, AppointmentWarehouseClient warehouses,
@Value("${baiyue.security.internal-service-token:}") String token) {
@@ -73,96 +72,12 @@ public class AppointmentSourceService {
public List<Source> resolveBatch(List<AppointmentCoverageService.SourceRef> requested) {
return AppointmentSourceService.this.resolveBatch(requested,true,this).stream().map(Resolution::requireSource).toList();
}
/** 同页只合并原料查询；来源规范化、重复与权限检查仍按每份需求独立执行。 */
public List<OrderGroup> resolveOrderGroups(List<List<AppointmentExternalSourceGateway.OrderRef>> groups) {
var canonical=new java.util.ArrayList<List<AppointmentCoverageService.SourceRef>>();
var errors=new java.util.ArrayList<RuntimeException>();
var original=new LinkedHashSet<AppointmentExternalSourceGateway.OrderRef>();
for(var group:groups) {
RuntimeException error=null;
if(group==null || group.isEmpty() || group.size()>100) error=new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"Firstleg需求缺少完整客户货源关系");
else if(group.stream().anyMatch(r->r==null || blank(r.customerId()) || r.orderId()==null || r.orderId()<=0)) error=bad("请选择完整的真实货源");
errors.add(error);if(error==null) original.addAll(group);
}
var aliases=new java.util.HashMap<AppointmentExternalSourceGateway.OrderRef,AppointmentExternalSourceGateway.Source>();
var originals=new java.util.ArrayList<>(original);
for(int start=0;start<originals.size();start+=100) {
for(var booking:external.resolveOrders(originals.subList(start,Math.min(start+100,originals.size())))) {
var key=new AppointmentExternalSourceGateway.OrderRef(booking.customerId(),booking.orderId());
if(aliases.put(key,booking)!=null) throw new IllegalStateException("外接原订单关联重复");
}
}
var bookingRefs=new LinkedHashSet<AppointmentExternalSourceGateway.SourceRef>();
var orderIds=new LinkedHashSet<Long>();
for(int i=0;i<groups.size();i++) {
if(errors.get(i)!=null) { canonical.add(List.of());continue; }
var refs=groups.get(i).stream().distinct().map(ref->{
var alias=aliases.get(ref);
return new AppointmentCoverageService.SourceRef(ref.customerId(),alias==null?"ORDER:"+ref.orderId():"BOOKING:"+alias.bookingNo(),null);
}).toList();
// 保留原 canonical 的数量校验，不能以全页去重掩盖单需求重复。
if(refs.size()!=groups.get(i).size()) errors.set(i,new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"来源键必须明确为ORDER或BOOKING"));
refs=refs.stream().distinct().toList();canonical.add(refs);
for(var ref:refs) if(ref.sourceKey().startsWith("ORDER:")) orderIds.add(orderId(ref.sourceKey()));
else bookingRefs.add(new AppointmentExternalSourceGateway.SourceRef(ref.customerId(),ref.sourceKey().substring(8)));
}
var bookingMap=new java.util.HashMap<String,AppointmentExternalSourceGateway.Source>();
var bookings=new java.util.ArrayList<>(bookingRefs);
for(int start=0;start<bookings.size();start+=100) for(var booking:external.resolve(bookings.subList(start,Math.min(start+100,bookings.size())))) {
if(bookingMap.put(booking.customerId()+"\n"+booking.bookingNo(),booking)!=null) throw bad("外接来源重复，请先核对");
if(booking.orderId()!=null) orderIds.add(booking.orderId());
}
var orderMap=new java.util.LinkedHashMap<Long,com.baiyue.lastleg.entity.LastlegOrder>();
var gates=new java.util.HashMap<String,Boolean>();
var destinations=new java.util.HashMap<Long,AppointmentCoverageMapper.Source>();
var ids=new java.util.ArrayList<>(orderIds);
for(int start=0;start<ids.size();start+=100) {
var rows=readRepository.orders(ids.subList(start,Math.min(start+100,ids.size())));
rows.forEach(row->{if(orderMap.put(row.getId(),row)!=null) throw new IllegalStateException("原订单来源重复");});
}
var containers=orderMap.values().stream().filter(o->"FIRST_LEG".equals(o.getOrderType())).map(com.baiyue.lastleg.entity.LastlegOrder::getSourceContainerBookingId).filter(Objects::nonNull).distinct().toList();
for(int start=0;start<containers.size();start+=100) readRepository.gates(containers.subList(start,Math.min(start+100,containers.size()))).forEach(g->{
if(gates.put(g.getContainerBookingId(),g.getNodeSequence()!=null && g.getNodeSequence()>=40)!=null) throw new IllegalStateException("原船节点门禁重复");
});
var transfers=orderMap.values().stream().filter(o->"WAREHOUSE_TRANSFER".equals(o.getOrderType())).map(o->new AppointmentCoverageMapper.OrderRef(o.getCustomerId(),o.getId())).toList();
for(int start=0;start<transfers.size();start+=100) readRepository.destinations(transfers.subList(start,Math.min(start+100,transfers.size()))).forEach(d->{
if(destinations.put(d.orderId(),d)!=null) throw new IllegalStateException("原目的地来源重复");
});
var result=new java.util.ArrayList<OrderGroup>();RuntimeException infrastructure=null;
for(int i=0;i<groups.size();i++) {
var refs=canonical.get(i);RuntimeException error=errors.get(i);var resolved=new java.util.ArrayList<Source>();
if(error==null) error=infrastructure;
if(error==null) try {
var keys=new java.util.HashSet<String>();var materialized=new java.util.HashSet<Long>();
for(var ref:refs) {
if(blank(ref.sourceKey()) || ref.sourceKey().length()>160 || blank(ref.customerId()) || ref.customerId().length()>128) throw bad("请选择完整的真实货源");
if(!keys.add(ref.sourceKey())) throw bad("同一来源货物不能重复选择");
}
var directory=catalog();
for(var ref:refs) {
var source=resolve(ref.sourceKey(),ref.customerId(),ref.destinationKey(),orderMap,bookingMap,gates,directory,destinations);
if(source.orderId()!=null && !materialized.add(source.orderId())) throw bad("同一货物存在重复来源，请保留原登记来源");
resolved.add(source);
}
} catch(RuntimeException failed) {
error=failed;
// 基础设施失败不重试、不变为空数据，延后到原需求访问位置抛出。
if(!(failed instanceof ResponseStatusException status) || status.getStatusCode().value()>=500) infrastructure=failed;
}
result.add(new OrderGroup(refs,List.copyOf(resolved),error));
}
return List.copyOf(result);
}
/** 读取失败不赋值，下次显式调用仍按原目录接口重读。 */
private List<AppointmentWarehouseClient.Warehouse> catalog() {
if(directory==null) directory=AppointmentSourceService.this.catalog();
return directory;
}
}
public record OrderGroup(List<AppointmentCoverageService.SourceRef> references,List<Source> sources,RuntimeException error) {
/** 业务前置错误仍在原需求解析位置抛出，由原边界决定是否保留未就绪需求。 */
public List<Source> requireSources() { if(error!=null) throw error;return sources; }
}
/** Read-only inspection keeps each source's prerequisite failure separate; infrastructure errors still escape. */
public record Resolution(AppointmentCoverageService.SourceRef reference,Source source,ResponseStatusException error) {
public Source requireSource() { if(error!=null) throw error; return source; }
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkFactAdapter.java
+16 -10
View File
@@ -4,6 +4,7 @@ import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.*;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@@ -60,14 +61,14 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
var requests=remote.subList(start,Math.min(start+200,remote.size())); timing.stage="FIRSTLEG"; requireToken();
try { var response=timing.measure(timing.firstleg,requests.size(),()->firstleg.facts(token,requests));
if(response==null || !response.successful()) throw unavailable("提柜及还空事实读取失败");
append(result,requests,response.data());
run(Stage.FACTS_VALIDATE,()->append(result,requests,response.data()));
} catch(RuntimeException error) { throw unavailable("Firstleg工作事实不可用",error); }
}
groups.forEach((provider,requests)->{
for(int start=0; start<requests.size(); start+=200) {
var batch=requests.subList(start,Math.min(start+200,requests.size()));
var statistics=timing.provider(provider);
try { append(result,batch,timing.measure(statistics,batch.size(),()->provider.read(batch))); }
try { var response=timing.measure(statistics,batch.size(),()->provider.read(batch));run(Stage.FACTS_VALIDATE,()->append(result,batch,response)); }
catch(RuntimeException error) { throw unavailable("预约/收货/运输事实不可用",error); }
}
});
@@ -76,6 +77,7 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
}
/** 一次 facts 调用的计时，不保存原来源或返回值，不跨调用缓存。 */
private static final class FactTiming {
private final Span prepare=span(Stage.FACTS_PREPARE);
private final long started=System.nanoTime();
private final CallStatistics firstleg=new CallStatistics();
private final Map<String,CallStatistics> providers=new LinkedHashMap<>();
@@ -92,20 +94,24 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
}
/** 只统计真正尝试的提供端调用；校验及组装耗时保留在 totalMs 中。 */
private <T> T measure(CallStatistics statistics,int requests,java.util.function.Supplier<T> operation) {
statistics.calls++;statistics.requests+=requests;long begin=System.nanoTime();
try { return operation.get(); }
finally { statistics.elapsed+=System.nanoTime()-begin; }
prepare.close();statistics.calls++;statistics.requests+=requests;long begin=System.nanoTime();
try(var span=span(switch(stage){case "FIRSTLEG"->Stage.FACTS_FIRSTLEG;case "AppointmentWorkReceivingProvider"->Stage.FACTS_RECEIVING;
case "AppointmentWorkDeliveryProvider"->Stage.FACTS_DELIVERY;case "AppointmentWorkTransportProvider"->Stage.FACTS_TRANSPORT;
case "AppointmentWorkDirectTransportProvider"->Stage.FACTS_DIRECT_TRANSPORT;default->Stage.FACTS_OTHER;})){
try{return operation.get();}catch(RuntimeException|Error error){span.failed();throw error;}
}finally{statistics.elapsed+=System.nanoTime()-begin;}
}
/** 成功慢调用或任意失败只输出一次汇总，日志故障不得遮蔽原异常。 */
private void finish(boolean success) {
if(!success)prepare.failed();prepare.close();
long totalMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime()-started);
if(success && totalMs<250) return;
try {
var summaries=new LinkedHashMap<String,String>();
providers.forEach((name,value)->summaries.put(name,value.summary()));
LOG.warn("Appointment work facts diagnostic: totalMs={}, firstlegCalls={}, firstlegRequests={}, firstlegMs={}, providers={}, failedStage={}",
totalMs,firstleg.calls,firstleg.requests,firstleg.millis(),summaries,success?"NONE":stage);
} catch(RuntimeException diagnosticFailure) { /* 仅忽略诊断故障，原事实错误仍沿原路径抛出。 */ }
LOG.warn("Appointment work facts diagnostic: totalMs={}, firstlegCalls={}, firstlegRequests={}, firstlegMs={}, providers={}, failedStage={}, requestId={}, traceId={}, scopeId={}",
totalMs,firstleg.calls,firstleg.requests,firstleg.millis(),summaries,success?"NONE":stage,org.slf4j.MDC.get("requestId"),org.slf4j.MDC.get("traceId"),currentScopeId());
} catch(Throwable diagnosticFailure) { /* 仅忽略诊断故障，原事实错误仍沿原路径抛出。 */ }
}
}
/** 仅包含数量和单调时钟耗时的累加器。 */
@@ -117,7 +123,7 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
private String summary() { return "calls="+calls+",requests="+requests+",ms="+millis(); }
}
/** 每次生成建立独立读取器；没有实际需求读取时不引入原业务依赖。 */
public RequirementReader openRequirements(List<RequirementInput> inputs) {
public RequirementReader openRequirements() {
return new RequirementReader() {
private RequirementReader reader;
/** 首次实际读取才取得原域作用域，保留每份需求的完整性校验。 */
@@ -125,7 +131,7 @@ public class AppointmentWorkFactAdapter implements AppointmentWorkFactGateway {
if(reader==null) {
var provider=requirements.getIfAvailable();
if(provider==null) throw unavailable("原预约需求与批次清单适配尚未接入，不能生成不完整工作清单");
reader=provider.openRequirements(inputs);
reader=provider.openRequirements();
}
var result=reader.read(source,effectiveFlow);
if(result==null || result.isEmpty()) throw unavailable("原预约需求清单缺失，不能以空清单完成路线生成");
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkFactGateway.java
+2 -3
View File
@@ -28,7 +28,7 @@ public interface AppointmentWorkFactGateway {
SourcePage discover(String after, int size);
ContainerSource container(String containerBookingId);
/** 建立本次原预约需求读取作用域，保持完整批次及明确替代关系。 */
RequirementReader openRequirements(List<RequirementInput> inputs);
RequirementReader openRequirements();
Map<String, Fact> facts(List<AppointmentWorkState> states);
Owner owner(String id, Actor actor);
Responsibility supplier(String id);
@@ -43,6 +43,5 @@ public interface AppointmentWorkFactGateway {
interface SupplierDirectory { Responsibility activeSupplier(String id); }
/** 单次生成显式复用目录，不跨请求保存来源事实。 */
interface RequirementReader { List<Requirement> read(ContainerSource source,Flow effectiveFlow); }
record RequirementInput(ContainerSource source,Flow flow) { }
interface RequirementProvider { RequirementReader openRequirements(List<RequirementInput> inputs); }
interface RequirementProvider { RequirementReader openRequirements(); }
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkQueryService.java
+32 -6
View File
@@ -9,6 +9,7 @@ import java.util.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@@ -26,10 +27,16 @@ public class AppointmentWorkQueryService {
Snapshot snapshot(Actor actor) {
return snapshot(actor,s->true,true);
}
/** 按原顺序构造完整快照，计时只分离读取、筛选、事实及投影。 */
private Snapshot snapshot(Actor actor,Predicate<AppointmentWorkState> selected,boolean includeNext) {
observeTransaction();var timing=span(Stage.QUERY_GUARD);
try {
String scopeVersion=store.requireInitialized();
timing.close();timing=span(Stage.QUERY_DISCOVER);
if(!Objects.equals(scopeVersion,facts.discover(null,1).scopeVersion())) throw conflict("已开船工作范围变化，请先完成服务端同步");
timing.close();timing=span(Stage.QUERY_STORE);
var all=store.all();
timing.close();timing=span(Stage.QUERY_SELECT);
var stateByKey=all.stream().collect(Collectors.toMap(AppointmentWorkState::getWorkItemKey,Function.identity()));
var selectedKeys=all.stream().filter(s->!s.isRetired() && selected.test(s)).map(AppointmentWorkState::getWorkItemKey).collect(Collectors.toSet());
var needed=new HashSet<>(selectedKeys);
@@ -41,12 +48,16 @@ public class AppointmentWorkQueryService {
}
}
var states=all.stream().filter(s->!s.isRetired() && needed.contains(s.getWorkItemKey())).toList();
timing.close();timing=span(Stage.QUERY_FACTS);
var current=facts.facts(states);
timing.close();timing=span(Stage.QUERY_PROJECT);
var visible=states.stream().filter(s->canRead(s,current.get(s.getWorkItemKey()),actor)).toList();
var items=visible.stream().map(s->project(s,current.get(s.getWorkItemKey()),actor,clock)).toList();
var byKey=items.stream().collect(Collectors.toMap(WorkItem::workItemKey,Function.identity()));
timing.close();timing=span(Stage.QUERY_NEXT);
return new Snapshot(all,current,items.stream().filter(i->selectedKeys.contains(i.workItemKey()))
.map(i->includeNext?withNext(i,next(i,byKey,stateByKey,new HashSet<>())):i).toList());
} catch(RuntimeException|Error error){timing.failed();throw error;} finally{timing.close();}
}
private Action next(WorkItem item,Map<String,WorkItem> items,Map<String,AppointmentWorkState> states,Set<String> visited) {
if(!visited.add(item.workItemKey()) || item.reviewRequired()) return item.nextAction();
@@ -73,23 +84,34 @@ public class AppointmentWorkQueryService {
}
return item.nextAction();
}
/** 完整快照之后单独统计汇总耗时，不改变待办筛选语义。 */
public Summary summary(Actor actor) {
var items=snapshot(actor,s->true,false).items().stream().filter(i->i.executionStatus()!=ExecutionStatus.COMPLETED).toList();
var snapshotItems=snapshot(actor,s->true,false).items();
return measure(Stage.QUERY_FINAL,()->{
var items=snapshotItems.stream().filter(i->i.executionStatus()!=ExecutionStatus.COMPLETED).toList();
return new Summary(items.size(),items.stream().filter(i->followToday(i,clock)).count(),items.stream().filter(i->!i.overdueReasons().isEmpty()).count(),items.stream().filter(i->i.calendarDate()==null).count(),items.stream().filter(WorkItem::reviewRequired).count());
});
}
/** 日历最终分组保持原顺序，单独记录快照后处理。 */
public CalendarResult calendar(LocalDate from,LocalDate to,Flow flow,Set<Type> types,Actor actor) {
if(from==null || to==null || to.isBefore(from) || ChronoUnit.DAYS.between(from,to)>=62) throw bad("日历范围须为1至62天");
if(types!=null && types.contains(Type.ROUTE)) throw bad("路线待确认需求请从全局待办查看");
var items=snapshot(actor,s->s.getType()!=Type.ROUTE && matches(s,flow,types,null),false).items().stream().filter(i->i.type()!=Type.ROUTE && i.calendarDate()!=null && !i.calendarDate().isBefore(from) && !i.calendarDate().isAfter(to))
var snapshotItems=snapshot(actor,s->s.getType()!=Type.ROUTE && matches(s,flow,types,null),false).items();
return measure(Stage.QUERY_FINAL,()->{
var items=snapshotItems.stream().filter(i->i.type()!=Type.ROUTE && i.calendarDate()!=null && !i.calendarDate().isBefore(from) && !i.calendarDate().isAfter(to))
.filter(i->flow==null || i.flow()==flow).filter(i->types==null || types.contains(i.type())).toList();
var groups=items.stream().collect(Collectors.groupingBy(WorkItem::calendarDate,TreeMap::new,Collectors.toList()));
var days=groups.entrySet().stream().map(e->new DayAggregate(e.getKey(),e.getValue().size(),e.getValue().stream().collect(Collectors.groupingBy(WorkItem::type,()->new EnumMap<>(Type.class),Collectors.counting())),counts(e.getValue()))).toList();
return new CalendarResult(from,to,items.size(),counts(items),days);
});
}
/** 保留事实核对之后的最终过滤排序分页，仅拆出其耗时。 */
public PageResult<WorkItem> items(View view,int page,int size,LocalDate date,Flow flow,Set<Type> types,String containerId,Bucket bucket,String keyword,Actor actor) {
if(page<1 || size<1 || size>100 || keyword!=null && keyword.length()>128) throw bad("分页或搜索条件无效");
var word=keyword==null?"":keyword.trim().toLowerCase(Locale.ROOT);
var items=snapshot(actor,s->matches(s,flow,types,containerId) && candidate(s,view),true).items().stream()
var snapshotItems=snapshot(actor,s->matches(s,flow,types,containerId) && candidate(s,view),true).items();
return measure(Stage.QUERY_FINAL,()->{
var items=snapshotItems.stream()
.filter(i->date!=null || i.executionStatus()!=ExecutionStatus.COMPLETED)
.filter(i->date==null || date.equals(i.calendarDate()))
.filter(i->flow==null || flow==i.flow()).filter(i->types==null || types.contains(i.type()))
@@ -100,6 +122,7 @@ public class AppointmentWorkQueryService {
.sorted(Comparator.comparing(WorkItem::followupDate,Comparator.nullsLast(Comparator.naturalOrder())).thenComparing(WorkItem::workItemKey)).toList();
long offset=(long)(page-1)*size; int start=(int)Math.min(offset,items.size());
return new PageResult<>(items.subList(start,Math.min(start+size,items.size())),page,size,items.size(),(items.size()+size-1L)/size);
});
}
public BatchWorkResolution resolveBatch(String batchKey,String customerId,Actor actor) {
if(batchKey==null || !batchKey.matches("(?:FBA_APPOINTMENT|DISPATCH):[1-9][0-9]*")) throw bad("请提供原业务返回的完整批次键");
@@ -117,10 +140,12 @@ public class AppointmentWorkQueryService {
var selected=candidates.get(0);
return new BatchWorkResolution(batchKey,selected.customerId(),detail(selected.workItemKey(),actor),candidates);
}
/** 保留详情完整快照与退休事实读取，历史查询单独观察。 */
public WorkDetail detail(String key,Actor actor) {
var snapshot=snapshot(actor); var state=store.get(key);
var snapshot=snapshot(actor); var state=measure(Stage.DETAIL_STORE,()->store.get(key));
var fact=snapshot.facts().get(key);
if(state.isRetired() && state.getType()!=Type.ROUTE) fact=facts.facts(List.of(state)).get(key);
if(state.isRetired() && state.getType()!=Type.ROUTE) fact=measure(Stage.DETAIL_FACTS,()->facts.facts(List.of(state))).get(key);
try(var timing=span(Stage.DETAIL_PROJECT)) {
if(!canRead(state,fact,actor)) throw forbidden("无权读取关联来源");
var item=state.isRetired()?project(state,fact,actor,clock):snapshot.items().stream().filter(i->i.workItemKey().equals(key)).findFirst().orElseThrow(()->missing("事项不存在"));
var relatedKeys=new HashSet<>(state.getPrerequisites()); relatedKeys.addAll(state.getReplacementKeys());
@@ -128,9 +153,10 @@ public class AppointmentWorkQueryService {
relatedKeys.addAll(effectiveKeys(relatedKeys,statesByKey));
snapshot.states().stream().filter(s->effectiveKeys(s.getPrerequisites(),statesByKey).contains(key)).map(AppointmentWorkState::getWorkItemKey).forEach(relatedKeys::add);
var refs=new ArrayList<>(state.getReferences()); if(fact!=null) refs.addAll(fact.references());
return new WorkDetail(item,snapshot.items().stream().filter(i->relatedKeys.contains(i.workItemKey())).toList(),refs.stream().distinct().toList(),store.history(key),
return new WorkDetail(item,snapshot.items().stream().filter(i->relatedKeys.contains(i.workItemKey())).toList(),refs.stream().distinct().toList(),measure(Stage.DETAIL_HISTORY,()->store.history(key)),
state.isReviewRequired()?new Review(state.getReviewVersion(),List.of(Objects.toString(state.getReviewReason(),"关联安排变化"))):null,
requirementOwned(state,fact)?new RequirementDetails(state.isBookingRequired(),Objects.toString(state.getReceivingRequirements(),""),state.getRequirementReason()):null,state.getReturnHandoff());
}
}
WorkItem current(AppointmentWorkState state,Fact fact,Actor actor) { return project(state,fact,actor,clock); }
private boolean matches(AppointmentWorkState state,Flow flow,Set<Type> types,String containerId) {
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkStore.java
+12 -7
View File
@@ -7,6 +7,7 @@ import org.slf4j.MDC;
import org.springframework.stereotype.Component;
import java.time.*;
import java.util.*;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@@ -21,13 +22,14 @@ public class AppointmentWorkStore {
public void invalidateRequirements() {
long started = System.nanoTime();
try {
if (mapper.invalidateRequirements() != 1) throw unavailable("工作事项同步守卫未初始化");
if (measure(Stage.GUARD_UPDATE,mapper::invalidateRequirements) != 1) throw unavailable("工作事项同步守卫未初始化");
} finally {
long elapsedMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
if (elapsedMs >= 250) {
// 语句耗时可能包含等锁或连接等待，不能仅凭日志认定具体锁根因；不记录请求业务载荷。
LOG.warn("Appointment work guard update slow: elapsedMs={}, requestId={}, traceId={}",
elapsedMs, MDC.get("requestId"), MDC.get("traceId"));
try { LOG.warn("Appointment work guard update slow: elapsedMs={}, requestId={}, traceId={}, scopeId={}",
elapsedMs, MDC.get("requestId"), MDC.get("traceId"), currentScopeId());
} catch(Throwable ignored) { /* 诊断输出失败不得覆盖原守卫结果或异常。 */ }
}
}
}
@@ -35,20 +37,23 @@ public class AppointmentWorkStore {
public AppointmentWorkMapper.Guard lock() {
long started = System.nanoTime();
try {
var guard = mapper.lockGuard();
observeTransaction();
var guard = measure(Stage.GUARD_LOCK,mapper::lockGuard);
if (guard == null) throw unavailable("工作事项DDL尚未集成");
return guard;
} finally {
long elapsedMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
if (elapsedMs >= 250) {
// 耗时包含连接及语句执行，单条日志不能区分具体等待原因，不记录业务载荷。
LOG.warn("Appointment work guard lock slow: elapsedMs={}, requestId={}, traceId={}",
elapsedMs, MDC.get("requestId"), MDC.get("traceId"));
try { LOG.warn("Appointment work guard lock slow: elapsedMs={}, requestId={}, traceId={}, scopeId={}",
elapsedMs, MDC.get("requestId"), MDC.get("traceId"), currentScopeId());
} catch(Throwable ignored) { /* 诊断输出失败不得覆盖原守卫结果或异常。 */ }
}
}
}
/** 完整性守卫保持原语义，读取耗时与事务完成只作诊断。 */
public String requireInitialized() {
var guard=mapper.guard();
observeTransaction();var guard=measure(Stage.GUARD_READ,mapper::guard);
if(guard==null || !guard.isInitialized()) throw unavailable("完整工作范围尚未完成同步，不能显示零项");
if(guard.getRequirementVersion()!=guard.getSynchronizedRequirementVersion()) throw conflict("预约批次或运输关联已变化，服务端正在协调完整工作清单");
return guard.getScopeVersion();
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/AppointmentWorkSyncService.java
+14 -15
View File
@@ -8,6 +8,7 @@ import java.time.*;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@@ -23,7 +24,7 @@ public class AppointmentWorkSyncService {
@Transactional
public SyncResult synchronize(SyncRequest request) {
if(request==null || request.size()<1 || request.size()>100) throw bad("同步批次须为1至100柜");
var timing=new SyncTiming();boolean success=false;
observeTransaction();var timing=new SyncTiming();boolean success=false;
try {
var guard=store.lock();
if(request.afterContainerBookingId()!=null && !Objects.equals(guard.getScanCursor(),request.afterContainerBookingId())) throw conflict("来源同步游标已变化，请从首批重新核对");
@@ -36,16 +37,7 @@ public class AppointmentWorkSyncService {
timing.next(SyncPhase.GENERATE);
// 保留全部历史与稳定键顺序；逐柜更新立即进入本次工作集，不跨事务缓存。
var known=workingSet();
var requirementInputs=page.containers().stream().filter(source->{
// 坏来源仍由下面原顺序校验并报错，不能在后页原料提示阶段抢先覆盖前柜错误。
try { validateSource(source);return true; } catch(RuntimeException invalid) { return false; }
}).map(source->{
var flow=known.values().stream().filter(s->source.containerBookingId().equals(s.getContainerBookingId()) || s.getContainers().stream().anyMatch(c->source.containerBookingId().equals(c.containerBookingId())))
.filter(s->s.getType()!=Type.ROUTE && !s.isRetired()).map(AppointmentWorkState::getFlow).findFirst().orElse(source.establishedFlow());
return new RequirementInput(source,flow);
}).filter(input->input.flow()!=Flow.UNKNOWN).toList();
// 提示仅用于原料批量读取，下面逐柜 existing/activeFlow 仍根据最新工作集决定。
var requirementReader=gateway.openRequirements(requirementInputs);
var requirementReader=gateway.openRequirements();
int created=0; var ids=new HashSet<String>();
for(var source:page.containers()) {
validateSource(source);
@@ -97,6 +89,8 @@ public class AppointmentWorkSyncService {
private enum SyncPhase { GUARD, DISCOVER, GENERATE, TRANSPORT_FACTS, RECONCILE, ACTIVE_STORE, ACTIVE_FACTS, OBSERVE_CHECKPOINT }
/** 仅统计本次方法执行，不含事务代理提交，不能当作精确锁持有时间。 */
private static final class SyncTiming {
private final Span body=span(Stage.SYNC_BODY);
private Span diagnosticPhase=span(Stage.SYNC_GUARD);
private final long started=System.nanoTime();
private final long[] elapsed=new long[SyncPhase.values().length];
private long phaseStarted=started;
@@ -106,19 +100,24 @@ public class AppointmentWorkSyncService {
/** 完成前一阶段后开始下一阶段；事实阶段包含紧邻的数据库回读。 */
private void next(SyncPhase next) {
long now=System.nanoTime();elapsed[phase.ordinal()]+=now-phaseStarted;phaseStarted=now;phase=next;
diagnosticPhase.close();diagnosticPhase=span(switch(next){
case GUARD->Stage.SYNC_GUARD;case DISCOVER->Stage.SYNC_DISCOVER;case GENERATE->Stage.SYNC_GENERATE;
case TRANSPORT_FACTS->Stage.SYNC_TRANSPORT_FACTS;case RECONCILE->Stage.SYNC_RECONCILE;
case ACTIVE_STORE->Stage.SYNC_ACTIVE_STORE;case ACTIVE_FACTS->Stage.SYNC_ACTIVE_FACTS;case OBSERVE_CHECKPOINT->Stage.SYNC_CHECKPOINT;});
}
/** 成功和失败共用一条慢调用摘要，未到达的计数保持为负一。 */
private void finish(boolean success) {
long now=System.nanoTime();elapsed[phase.ordinal()]+=now-phaseStarted;
if(!success){diagnosticPhase.failed();body.failed();}diagnosticPhase.close();body.close();
long totalMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(now-started);
if(totalMs<250) return;
try {
// 旧 activeFactsMs 仍包含工作表读取、解析及适配；新增明细不能改变旧指标口径。
long activeMs=java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed[SyncPhase.ACTIVE_STORE.ordinal()]+elapsed[SyncPhase.ACTIVE_FACTS.ordinal()]);
LOG.warn("Appointment work sync slow: totalMs={}, guardMs={}, discoverMs={}, generateMs={}, transportFactsMs={}, reconcileMs={}, activeFactsMs={}, activeStoreMs={}, activeAdapterMs={}, observeCheckpointMs={}, pageContainers={}, transportFactsCount={}, activeFactsCount={}, complete={}, failedStage={}",
LOG.warn("Appointment work sync slow: totalMs={}, guardMs={}, discoverMs={}, generateMs={}, transportFactsMs={}, reconcileMs={}, activeFactsMs={}, activeStoreMs={}, activeAdapterMs={}, observeCheckpointMs={}, pageContainers={}, transportFactsCount={}, activeFactsCount={}, complete={}, failedStage={}, requestId={}, traceId={}, scopeId={}",
totalMs,millis(SyncPhase.GUARD),millis(SyncPhase.DISCOVER),millis(SyncPhase.GENERATE),millis(SyncPhase.TRANSPORT_FACTS),millis(SyncPhase.RECONCILE),activeMs,millis(SyncPhase.ACTIVE_STORE),millis(SyncPhase.ACTIVE_FACTS),millis(SyncPhase.OBSERVE_CHECKPOINT),
pageContainers,transportFactsCount,activeFactsCount,complete,success?"NONE":phase.name());
} catch(RuntimeException diagnosticFailure) { /* 诊断失败不能替换原业务异常或阻断提交。 */ }
pageContainers,transportFactsCount,activeFactsCount,complete,success?"NONE":phase.name(),org.slf4j.MDC.get("requestId"),org.slf4j.MDC.get("traceId"),currentScopeId());
} catch(Throwable diagnosticFailure) { /* 诊断失败不能替换原业务异常或阻断提交。 */ }
}
/** 阶段仅输出整数毫秒，不携带业务标识。 */
private long millis(SyncPhase value) { return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(elapsed[value.ordinal()]); }
@@ -140,7 +139,7 @@ public class AppointmentWorkSyncService {
}
/** 命令入口仍在调用方守卫事务中即时加载工作集。 */
List<String> generate(ContainerSource source,Flow flow,List<AppointmentWorkState> existing,Actor actor) {
return generate(source,flow,existing,actor,workingSet(),gateway.openRequirements(List.of(new RequirementInput(source,flow))));
return generate(source,flow,existing,actor,workingSet(),gateway.openRequirements());
}
/** 与持久层全量读取保持稳定键顺序，并保留退役事项及原读取上限。 */
private Map<String,AppointmentWorkState> workingSet() {
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/diagnostics/AppointmentWorkDataSourcePostProcessor.java
+35
View File
@@ -0,0 +1,35 @@
package com.baiyue.lastleg.appointmentwork.diagnostics;
import com.zaxxer.hikari.HikariDataSource;
import org.aopalliance.intercept.MethodInterceptor;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.config.BeanPostProcessor;
import javax.sql.DataSource;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
/** 仅观察原Hikari getConnection；同池同连接，unwrap保持容器bean self契约而非隐藏target身份。 */
final class AppointmentWorkDataSourcePostProcessor implements BeanPostProcessor {
/** 不识别或已有代理的数据源保持原对象，明确记录未覆盖而不主动探测连接。 */
@Override public Object postProcessAfterInitialization(Object bean,String name) {
if(!(bean instanceof DataSource))return bean;
if(!(bean instanceof HikariDataSource) || AopUtils.isAopProxy(bean)) {
warnUnavailable();return bean;
}
try {
var factory=new ProxyFactory(bean);factory.setProxyTargetClass(true);
factory.addAdvice((MethodInterceptor)invocation->{
if(!invocation.getMethod().getName().equals("getConnection") || !active())return invocation.proceed();
try(var span=span(Stage.CONNECTION_ACQUIRE)) {
try{return invocation.proceed();}catch(Throwable error){span.failed();throw error;}
}
});
return factory.getProxy(bean.getClass().getClassLoader());
} catch(RuntimeException failure) { warnUnavailable();return bean; }
}
/** 不输出 Bean 名、数据源属性或异常内容。 */
private static void warnUnavailable() {
try{org.slf4j.LoggerFactory.getLogger(AppointmentWorkDataSourcePostProcessor.class).warn("Appointment connection diagnostic unavailable: unsupported or already proxied datasource");}
catch(Throwable ignored){ /* 日志故障不改变 Bean 注册。 */ }
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/diagnostics/AppointmentWorkDiagnosticFilter.java
+39
View File
@@ -0,0 +1,39 @@
package com.baiyue.lastleg.appointmentwork.diagnostics;
import jakarta.servlet.*;
import jakarta.servlet.http.*;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import java.io.IOException;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
/** 仅测量服务端同步 filter dispatch，不代表客户端总耗时或异步响应完成时间。 */
@Component
@Order(Ordered.HIGHEST_PRECEDENCE+10)
public class AppointmentWorkDiagnosticFilter extends OncePerRequestFilter {
private static final String BASE="/api/platform/v1/last-leg/appointment-work";
/** 仅识别固定模板；路径参数、query string和载荷不进入上下文。 */
static Route route(String method,String path){
if("GET".equals(method)){
if((BASE+"/summary").equals(path))return Route.SUMMARY;
if((BASE+"/calendar").equals(path))return Route.CALENDAR;
if((BASE+"/items").equals(path))return Route.ITEMS;
if((BASE+"/items/resolve").equals(path))return Route.RESOLVE;
if(path!=null && path.matches(BASE+"/items/[^/]+"))return Route.DETAIL;
}
if("POST".equals(method) && path!=null && path.matches(BASE+"/items/[^/]+/commands"))return Route.COMMAND;
return null;
}
/** 不采集其他端点，也不为诊断增加权限判断。 */
@Override protected boolean shouldNotFilter(HttpServletRequest request){return route(request.getMethod(),request.getRequestURI())==null;}
/** 创建及清理线程作用域；不覆写上游相关标识、状态、响应或异常。 */
@Override protected void doFilterInternal(HttpServletRequest request,HttpServletResponse response,FilterChain chain)throws ServletException,IOException{
Scope scope=null;
try{scope=open(Origin.HTTP,route(request.getMethod(),request.getRequestURI()));}catch(RuntimeException ignored){ /* 无诊断仍执行原请求。 */ }
try{chain.doFilter(request,response);}
catch(ServletException|IOException|RuntimeException|Error error){if(scope!=null)scope.failed();throw error;}
finally{if(scope!=null){try{scope.status(response.getStatus());}catch(Throwable ignored){ /* 响应诊断读取失败不替换原异常。 */ }finally{scope.close();}}}
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/diagnostics/AppointmentWorkDiagnostics.java
+128
View File
@@ -0,0 +1,128 @@
package com.baiyue.lastleg.appointmentwork.diagnostics;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import java.util.*;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
/** 仅在当前线程的预约调用内累加固定计时桶，不保存业务参数或跨请求状态。 */
public final class AppointmentWorkDiagnostics {
public enum Origin { HTTP, QUERY, COMMAND, SYNC }
public enum Route { INTERNAL, SUMMARY, CALENDAR, ITEMS, RESOLVE, DETAIL, COMMAND }
public enum Stage {
TRANSACTION_ADVISOR, QUERY_GUARD, QUERY_DISCOVER, QUERY_STORE, QUERY_SELECT, QUERY_FACTS,
QUERY_PROJECT, QUERY_NEXT, QUERY_FINAL, DETAIL_STORE, DETAIL_FACTS, DETAIL_PROJECT, DETAIL_HISTORY,
GUARD_LOCK, GUARD_READ, GUARD_UPDATE, FACTS_PREPARE, FACTS_FIRSTLEG, FACTS_RECEIVING,
FACTS_DELIVERY, FACTS_TRANSPORT, FACTS_DIRECT_TRANSPORT, FACTS_OTHER, FACTS_VALIDATE,
DELIVERY_SOURCE_MAP, DELIVERY_CARGO, DELIVERY_SNAPSHOTS, DELIVERY_ROWS, DELIVERY_RECEIVING,
DELIVERY_IMPACTS, DELIVERY_ASSEMBLY, SYNC_BODY, SYNC_GUARD, SYNC_DISCOVER, SYNC_GENERATE,
SYNC_TRANSPORT_FACTS, SYNC_RECONCILE, SYNC_ACTIVE_STORE, SYNC_ACTIVE_FACTS, SYNC_CHECKPOINT, CONNECTION_ACQUIRE,
DB_EXECUTOR_QUERY, DB_EXECUTOR_UPDATE, DB_PREPARE, DB_STATEMENT_QUERY, DB_STATEMENT_UPDATE, DB_RESULT_MAPPING
}
public enum StatementBucket { NONE, GUARD_LOCK, GUARD_READ, GUARD_UPDATE, WORK_STATE, WORK_HISTORY, DELIVERY, RECEIVING, OTHER }
public record Metric(long calls,long inclusiveNanos,long selfNanos,long failures,long firstOffsetNanos,long lastOffsetNanos) { }
public record Snapshot(String scopeId,String parentScopeId,Origin origin,Route route,String requestId,String traceId,
long elapsedNanos,boolean failed,int status,Map<String,Metric> metrics,boolean transactionObserved,int completionStatus,long completionOffsetNanos,long startedNanos) { }
private static final org.slf4j.Logger LOG=LoggerFactory.getLogger(AppointmentWorkDiagnostics.class);
private static final ThreadLocal<Scope> CURRENT=new ThreadLocal<>();
private static final Span NO_SPAN=new Span(null,null,null,0);
/** 诊断工具无需实例化。 */
private AppointmentWorkDiagnostics() { }
/** 新建独立作用域；关闭时恢复调用方原作用域，不改变 MDC。 */
public static Scope open(Origin origin,Route route) { return new Scope(origin,route,System::nanoTime); }
/** 返回纯诊断标识，缺少作用域时明确为空，不生成业务关联键。 */
public static String currentScopeId() { var scope=CURRENT.get();return scope==null?null:scope.id; }
/** 无作用域时所有阶段计时均为空操作。 */
public static boolean active() { return CURRENT.get()!=null; }
/** 只监听既有事务完成，不开启事务或改变其同步、隔离级别及连接属性。 */
public static void observeTransaction() {
try {
var scope=CURRENT.get();
if(scope==null || scope.transactionObserved || !org.springframework.transaction.support.TransactionSynchronizationManager.isSynchronizationActive()
|| !org.springframework.transaction.support.TransactionSynchronizationManager.isActualTransactionActive())return;
org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization(new org.springframework.transaction.support.TransactionSynchronization(){
/** 回调发生在原事务完成后，仅记录相对时刻；不声称这是数据库精确释放锁时刻。 */
@Override public void afterCompletion(int status){try{scope.completionStatus=status;scope.completionOffsetNanos=Math.max(0,scope.clock.getAsLong()-scope.started);}catch(Throwable ignored){}}
});
scope.transactionObserved=true;
}catch(Throwable ignored){ /* 诊断监听失败不能改变事务。 */ }
}
/** 将可信相关标识限制为诊断字符集；任意文本不进入日志。 */
private static String correlation(String value) { return value!=null && value.matches("[A-Za-z0-9_-]{1,80}")?value:null; }
/** 固定阶段开始计时；最多保留64层，避免异常递归造成诊断状态膨胀。 */
public static Span span(Stage stage) { return span(stage,StatementBucket.NONE); }
/** 数据库阶段只接受固定语句桶，不记录 SQL、参数或动态 Mapper 名称。 */
public static Span span(Stage stage,StatementBucket bucket) {
try {
var scope=CURRENT.get();if(scope==null || scope.stack.size()>=64)return NO_SPAN;
var span=new Span(scope,stage,bucket,scope.clock.getAsLong());scope.stack.push(span);return span;
} catch(RuntimeException ignored) { return NO_SPAN; }
}
/** 保留返回对象和原异常；测量只包围原有调用一次。 */
public static <T> T measure(Stage stage,Supplier<T> operation) {
try(var span=span(stage)) { try{return operation.get();}catch(RuntimeException|Error error){span.failed();throw error;} }
}
/** 无返回值调用沿相同路径执行，不增加重试或业务判断。 */
public static void run(Stage stage,Runnable operation) { measure(stage,()->{operation.run();return null;}); }
/** 读取当前统计仅用于固定诊断摘要和隔离验证。 */
static Snapshot currentSnapshot() { var scope=CURRENT.get();return scope==null?null:scope.snapshot(); }
public static final class Scope implements AutoCloseable {
private final Scope previous;
private final String id=UUID.randomUUID().toString();
private final Origin origin;private final Route route;private final LongSupplier clock;
private final String requestId,traceId;private final long started;
private final Deque<Span> stack=new ArrayDeque<>();
private final Map<String,long[]> totals=new LinkedHashMap<>();
private boolean failed,closed,transactionObserved;private int status,completionStatus=-1;private long completionOffsetNanos=-1;
/** 包内时钟注入仅供确定性验证，不修改全局时钟。 */
Scope(Origin origin,Route route,LongSupplier clock) {
this.previous=CURRENT.get();this.origin=origin;this.route=route;this.clock=clock;this.started=clock.getAsLong();
this.requestId=correlation(MDC.get("requestId"));this.traceId=correlation(MDC.get("traceId"));CURRENT.set(this);
}
/** 只标记失败，不保存异常类型以外的业务内容。 */
public void failed() { failed=true; }
/** HTTP状态由容器响应读取，业务code不从载荷解析。 */
public void status(int status) { this.status=status; }
/** 快照为累计已结束阶段；self仅扣同scope直接子段，独立子scope由parentScopeId关联且不可叠加。 */
public Snapshot snapshot() {
var metrics=new LinkedHashMap<String,Metric>();totals.forEach((key,v)->metrics.put(key,new Metric(v[0],v[1],v[2],v[3],v[4],v[5])));
return new Snapshot(id,previous==null?null:previous.id,origin,route,requestId,traceId,Math.max(0,clock.getAsLong()-started),failed,status,Collections.unmodifiableMap(metrics),transactionObserved,completionStatus,completionOffsetNanos,started);
}
/** 只有慢调用或失败输出汇总；清理优先，日志异常不得覆盖原业务异常。 */
@Override public void close() {
if(closed)return;closed=true;
try {
if(CURRENT.get()==this){if(previous==null)CURRENT.remove();else CURRENT.set(previous);}
var s=snapshot();if(!s.failed() && s.status()<500 && s.elapsedNanos()<250_000_000L)return;
// startedNanos及offset只可在同一JVM进程比较，不是数据库锁持有者或跨进程时钟。
LOG.warn("Appointment attribution: scopeId={}, parentScopeId={}, origin={}, route={}, requestId={}, traceId={}, startedNanos={}, elapsedNs={}, failed={}, httpStatus={}, connectionCoverage={}, transactionObserved={}, completionStatus={}, completionOffsetNs={}, boundary=SERVER_DISPATCH_OR_ADVISOR, stages={}",
s.scopeId(),s.parentScopeId(),s.origin(),s.route(),s.requestId(),s.traceId(),s.startedNanos(),s.elapsedNanos(),s.failed(),s.status(),
s.metrics().containsKey(Stage.CONNECTION_ACQUIRE.name())?"OBSERVED":"NOT_OBSERVED",s.transactionObserved(),s.completionStatus(),s.completionOffsetNanos(),s.metrics());
} catch(Throwable ignored) { /* 诊断设施故障不影响返回、原异常或线程清理。 */ }
}
}
public static final class Span implements AutoCloseable {
private final Scope scope;private final Stage stage;private final StatementBucket bucket;private final long started;
private long children;private boolean failed,closed;
/** 保存常量阶段，不引用任何业务对象。 */
private Span(Scope scope,Stage stage,StatementBucket bucket,long started){this.scope=scope;this.stage=stage;this.bucket=bucket;this.started=started;}
/** 失败计数只作为耗时解释，不改变原异常。 */
public void failed(){if(scope!=null){failed=true;scope.failed=true;}}
/** 每个阶段只结束一次；父段self扣除子段inclusive，不能把两种口径相加。 */
@Override public void close(){
if(scope==null || closed)return;closed=true;
try{
long elapsed=Math.max(0,scope.clock.getAsLong()-started);
if(scope.stack.peek()!=this)return;
scope.stack.pop();if(!scope.stack.isEmpty())scope.stack.peek().children+=elapsed;
var key=stage.name()+(bucket==StatementBucket.NONE?"":":"+bucket.name());
var v=scope.totals.computeIfAbsent(key,ignored->new long[]{0,0,0,0,Math.max(0,started-scope.started),0});v[0]++;v[1]+=elapsed;v[2]+=Math.max(0,elapsed-children);if(failed)v[3]++;
v[4]=Math.min(v[4],Math.max(0,started-scope.started));v[5]=Math.max(v[5],Math.max(0,started+elapsed-scope.started));
}catch(Throwable ignored){ /* 诊断计数故障不能替换原结果。 */ }
}
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/diagnostics/AppointmentWorkDiagnosticsConfiguration.java
+47
View File
@@ -0,0 +1,47 @@
package com.baiyue.lastleg.appointmentwork.diagnostics;
import com.baiyue.lastleg.appointmentwork.*;
import org.aopalliance.intercept.MethodInterceptor;
import org.springframework.aop.Advisor;
import org.springframework.aop.support.DefaultPointcutAdvisor;
import org.springframework.aop.support.StaticMethodMatcherPointcut;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Role;
import org.springframework.core.Ordered;
import java.lang.reflect.Method;
import java.util.Set;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
/** 使用现有 Spring 代理设施观察事务边界，不新增 AspectJ 或改变事务属性。 */
@Configuration(proxyBeanMethods=false)
public class AppointmentWorkDiagnosticsConfiguration {
/** 基础设施角色确保事务自动代理器纳入此 Advisor；顺序早于默认事务 Advisor。 */
@Bean @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public Advisor appointmentWorkDiagnosticAdvisor() {
var pointcut=new StaticMethodMatcherPointcut() {
/** 仅匹配预约直接入口，不拦截其他模块或内部辅助方法。 */
@Override public boolean matches(Method method,Class<?> targetClass) {
return AppointmentWorkSyncService.class.isAssignableFrom(targetClass) && method.getName().equals("synchronize")
|| AppointmentWorkCommandService.class.isAssignableFrom(targetClass) && method.getName().equals("execute")
|| AppointmentWorkQueryService.class.isAssignableFrom(targetClass) && Set.of("summary","calendar","items","resolveBatch","detail").contains(method.getName());
}
};
MethodInterceptor interceptor=invocation->{
boolean sync=invocation.getThis() instanceof AppointmentWorkSyncService;
Origin origin=sync?Origin.SYNC:invocation.getThis() instanceof AppointmentWorkCommandService?Origin.COMMAND:Origin.QUERY;
Scope owned=null;
try { if(sync || !active())owned=open(origin,Route.INTERNAL); } catch(RuntimeException ignored) { /* 无法计时仍执行原方法。 */ }
try(var span=span(Stage.TRANSACTION_ADVISOR)) {
try{return invocation.proceed();}catch(Throwable error){span.failed();if(owned!=null)owned.failed();throw error;}
} finally { if(owned!=null)owned.close(); }
};
var advisor=new DefaultPointcutAdvisor(pointcut,interceptor);advisor.setOrder(Ordered.HIGHEST_PRECEDENCE+100);return advisor;
}
/** 静态 Bean 避免提早实例化业务依赖，只对既有 Hikari 实例增加委托代理。 */
@Bean public static BeanPostProcessor appointmentWorkDataSourceDiagnostics() { return new AppointmentWorkDataSourcePostProcessor(); }
/** MyBatis 按现有插件入口注册，不注册 Mapper、执行 SQL 或改变缓存配置。 */
@Bean public AppointmentWorkMyBatisDiagnostics appointmentWorkMyBatisDiagnostics() { return new AppointmentWorkMyBatisDiagnostics(); }
}
lastleg-service/src/main/java/com/baiyue/lastleg/appointmentwork/diagnostics/AppointmentWorkMyBatisDiagnostics.java
+51
View File
@@ -0,0 +1,51 @@
package com.baiyue.lastleg.appointmentwork.diagnostics;
import org.apache.ibatis.executor.Executor;
import org.apache.ibatis.executor.resultset.ResultSetHandler;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.session.*;
import java.sql.*;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
/** query 为含映射的 inclusive 调用；结果读取另记子段，不推断为纯服务器锁等待。 */
@Intercepts({
@Signature(type=Executor.class,method="query",args={MappedStatement.class,Object.class,RowBounds.class,ResultHandler.class}),
@Signature(type=Executor.class,method="query",args={MappedStatement.class,Object.class,RowBounds.class,ResultHandler.class,CacheKey.class,BoundSql.class}),
@Signature(type=Executor.class,method="update",args={MappedStatement.class,Object.class}),
@Signature(type=StatementHandler.class,method="prepare",args={Connection.class,Integer.class}),
@Signature(type=StatementHandler.class,method="query",args={Statement.class,ResultHandler.class}),
@Signature(type=StatementHandler.class,method="update",args={Statement.class}),
@Signature(type=ResultSetHandler.class,method="handleResultSets",args={Statement.class})
})
public final class AppointmentWorkMyBatisDiagnostics implements Interceptor {
private static final ThreadLocal<StatementBucket> BUCKET=new ThreadLocal<>();
/** 所有调用只 proceed 一次，异常原样返回；固定桶上下文在 finally 恢复。 */
@Override public Object intercept(Invocation invocation)throws Throwable {
if(!active())return invocation.proceed();
var previous=BUCKET.get();boolean executor=invocation.getTarget() instanceof Executor;
var bucket=previous==null?StatementBucket.OTHER:previous;
if(executor){bucket=bucket(((MappedStatement)invocation.getArgs()[0]).getId());BUCKET.set(bucket);}
Stage stage=executor?(invocation.getMethod().getName().equals("query")?Stage.DB_EXECUTOR_QUERY:Stage.DB_EXECUTOR_UPDATE)
:invocation.getTarget() instanceof ResultSetHandler?Stage.DB_RESULT_MAPPING
:switch(invocation.getMethod().getName()){case "prepare"->Stage.DB_PREPARE;case "query"->Stage.DB_STATEMENT_QUERY;default->Stage.DB_STATEMENT_UPDATE;};
try(var span=span(stage,bucket)){
try{return invocation.proceed();}catch(Throwable error){span.failed();throw error;}
}finally{if(executor){if(previous==null)BUCKET.remove();else BUCKET.set(previous);}}
}
/** 只有明确白名单语句有独立桶，其他名字一律合入 OTHER，绝不打印原 ID。 */
private static StatementBucket bucket(String id){
return switch(id){
case "com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.lockGuard"->StatementBucket.GUARD_LOCK;
case "com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.guard"->StatementBucket.GUARD_READ;
case "com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.invalidateRequirements","com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.saveGuard"->StatementBucket.GUARD_UPDATE;
case "com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.all","com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.get"->StatementBucket.WORK_STATE;
case "com.baiyue.lastleg.appointmentwork.AppointmentWorkMapper.selectHistory"->StatementBucket.WORK_HISTORY;
case "com.baiyue.lastleg.inboundplan.work.AppointmentWorkDeliveryMapper.batches","com.baiyue.lastleg.inboundplan.work.AppointmentWorkDeliveryMapper.customer","com.baiyue.lastleg.inboundplan.work.AppointmentWorkDeliveryMapper.privateBatch"->StatementBucket.DELIVERY;
default->StatementBucket.OTHER;
};
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/InboundFbaAppointmentQueryMapper.java
-22
View File
@@ -28,28 +28,6 @@ public interface InboundFbaAppointmentQueryMapper {
" ORDER BY appointment_id DESC LIMIT 1001</script>"})
List<Long> relatedBatchIds(@Param("sources") List<com.baiyue.lastleg.appointmentplanning.AppointmentCoverageService.SourceRef> sources);
record RequirementSource(int groupOrdinal,String customerId,String sourceKey,String destinationKey) { }
record RequirementBatch(int groupOrdinal,long appointmentId) { }
/** 按原需求组返回历史归属，失活明细也保留；每组第1001条用于检测超限。 */
@Select("""
<script>
SELECT group_ordinal,appointment_id FROM (
SELECT matched.*,ROW_NUMBER() OVER(PARTITION BY group_ordinal ORDER BY appointment_id DESC) work_row FROM (
SELECT DISTINCT requested.group_ordinal,i.appointment_id
FROM ll_inbound_fba_appointment_item i JOIN (
<foreach collection='sources' item='s' separator=' UNION ALL '>
SELECT CAST(#{s.groupOrdinal} AS SIGNED) group_ordinal,
CAST(#{s.customerId} AS CHAR(128)) customer_id,CAST(#{s.sourceKey} AS CHAR(160)) source_key,
CAST(#{s.destinationKey} AS CHAR(160)) destination_key
</foreach>
) requested ON requested.customer_id=i.customer_id AND requested.source_key=i.source_key
AND requested.destination_key=CONCAT('FBA:',i.fba_warehouse_code)
) matched
) ranked WHERE work_row &lt;= 1001 ORDER BY group_ordinal,appointment_id DESC
</script>
""")
List<RequirementBatch> requirementBatches(@Param("sources") List<RequirementSource> sources);
@Select("<script>SELECT COUNT(*) FROM ll_order o "+com.baiyue.lastleg.appointmentplanning.AppointmentCoverageMapper.TRANSFER_JOIN+
" WHERE o.id IN <foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach> AND ("+
com.baiyue.lastleg.appointmentplanning.AppointmentCoverageMapper.SHIPMENT_MISSING+") </script>")
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundFbaAppointmentService.java
-37
View File
@@ -34,7 +34,6 @@ public class InboundFbaAppointmentService {
@Resource private com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore receiving;
@Resource private com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentWorkPort receivingWork;
@Resource private com.baiyue.lastleg.inboundplan.work.AppointmentWorkAppointmentChanges workChanges;
@Resource private com.baiyue.lastleg.inboundplan.work.AppointmentRequirementReadRepository requirementReads;
@Transactional(readOnly=true)
public List<FbaAppointmentResponse> calendar(LocalDate from,LocalDate to) {
@@ -112,42 +111,6 @@ public class InboundFbaAppointmentService {
if(ids.size()>1000) throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"关联批次超过读取范围，请缩小原来源范围后继续核对");
return ids.isEmpty()?List.of():responses(appointmentMapper.selectBatchIds(ids));
}
/** 同页原料只读一次，历史归属按原需求组保留；响应解析延后至原访问顺序。 */
public java.util.function.IntFunction<List<FbaAppointmentResponse>> prepareRelatedBatchSnapshots(List<List<AppointmentCoverageService.SourceRef>> groups) {
var errors=new HashMap<Integer,RuntimeException>();
var requested=new ArrayList<InboundFbaAppointmentQueryMapper.RequirementSource>();
for(int i=0;i<groups.size();i++) {
var refs=groups.get(i);
if(refs==null || refs.isEmpty()) continue;
if(refs.size()>100 || refs.stream().anyMatch(ref->ref==null || clean(ref.customerId())==null || clean(ref.sourceKey())==null || clean(ref.destinationKey())==null)) {
errors.put(i,bad("每次最多读取100个完整原来源的关联批次"));continue;
}
final int group=i;
refs.stream().distinct().forEach(ref->requested.add(new InboundFbaAppointmentQueryMapper.RequirementSource(group,ref.customerId(),ref.sourceKey(),ref.destinationKey())));
}
var groupIds=new HashMap<Integer,Set<Long>>();
for(int start=0;start<requested.size();start+=100) {
for(var match:requirementReads.history(requested.subList(start,Math.min(start+100,requested.size())))) {
if(match.groupOrdinal()<0 || match.groupOrdinal()>=groups.size()) throw new IllegalStateException("历史关联返回了请求外需求组");
groupIds.computeIfAbsent(match.groupOrdinal(),ignored->new TreeSet<>()).add(match.appointmentId());
}
}
groupIds.forEach((group,ids)->{if(ids.size()>1000) errors.put(group,new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"关联批次超过读取范围，请缩小原来源范围后继续核对"));});
var ids=groupIds.entrySet().stream().filter(e->!errors.containsKey(e.getKey())).flatMap(e->e.getValue().stream()).distinct().sorted().toList();
var rows=new LinkedHashMap<Long,InboundFbaAppointment>();
var items=new HashMap<Long,List<InboundFbaAppointmentItem>>();
for(int start=0;start<ids.size();start+=200) {
var batch=ids.subList(start,Math.min(start+200,ids.size()));
requirementReads.appointments(batch).forEach(row->rows.put(row.getId(),row));
requirementReads.items(batch).forEach(item->items.computeIfAbsent(item.getAppointmentId(),ignored->new ArrayList<>()).add(item));
}
return group->{
if(group<0 || group>=groups.size()) throw new IllegalArgumentException("需求组不属于本次读取范围");
if(errors.containsKey(group)) throw errors.get(group);
var selected=groupIds.getOrDefault(group,Set.of());
return rows.values().stream().filter(row->selected.contains(row.getId())).map(row->response(row,items.getOrDefault(row.getId(),List.of()))).toList();
};
}
@Transactional(readOnly=true)
public FbaAppointmentBatchContext batchContext(String key) {
final long id;
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/work/AppointmentRequirementReadRepository.java
-28
View File
@@ -1,28 +0,0 @@
package com.baiyue.lastleg.inboundplan.work;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baiyue.lastleg.inboundplan.entity.*;
import com.baiyue.lastleg.inboundplan.mapper.*;
import org.springframework.stereotype.Repository;
import java.util.List;
/** 需求生成的本域原料读取，历史归属不能从活动响应明细反推。 */
@Repository
public class AppointmentRequirementReadRepository {
private final AppointmentWorkSourceMapper mapping;
private final InboundFbaAppointmentQueryMapper queries;
private final InboundFbaAppointmentMapper appointments;
private final InboundFbaAppointmentItemMapper items;
/** 保留本域 Mapper 与客户来源查询边界。 */
public AppointmentRequirementReadRepository(AppointmentWorkSourceMapper mapping,InboundFbaAppointmentQueryMapper queries,InboundFbaAppointmentMapper appointments,InboundFbaAppointmentItemMapper items) {
this.mapping=mapping;this.queries=queries;this.appointments=appointments;this.items=items;
}
/** 每柜独立上限由既有窗口查询保持。 */
public List<AppointmentWorkSourceMapper.ContainerSource> sources(List<String> containers) { return containers.isEmpty()?List.of():mapping.containerSourcesBatch(containers); }
/** 分块只控制单条 SQL 输入，业务服务仍按需求组完整合并及核对上限。 */
public List<InboundFbaAppointmentQueryMapper.RequirementBatch> history(List<InboundFbaAppointmentQueryMapper.RequirementSource> sources) { return sources.isEmpty()?List.of():queries.requirementBatches(sources); }
/** 读取已确认关联的预约主记录。 */
public List<InboundFbaAppointment> appointments(List<Long> ids) { return ids.isEmpty()?List.of():appointments.selectBatchIds(ids); }
/** 包括失活明细以维持取消与替代历史；响应过滤由业务服务执行。 */
public List<InboundFbaAppointmentItem> items(List<Long> ids) { return ids.isEmpty()?List.of():items.selectList(new LambdaQueryWrapper<InboundFbaAppointmentItem>().in(InboundFbaAppointmentItem::getAppointmentId,ids).orderByAsc(InboundFbaAppointmentItem::getId)); }
}
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/work/AppointmentWorkDeliveryProvider.java
+9 -7
View File
@@ -8,6 +8,7 @@ import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore;
import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.BatchReceiving;
import com.baiyue.lastleg.security.OverseasPermissions;
import org.springframework.stereotype.Component;
import static com.baiyue.lastleg.appointmentwork.diagnostics.AppointmentWorkDiagnostics.*;
import java.util.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
@@ -26,11 +27,12 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
this.appointments=appointments; this.cargo=cargo; this.commitments=commitments; this.deliveries=deliveries; this.sourceMapping=sourceMapping;this.receiving=receiving;
}
public boolean supports(Type type,Flow flow) { return type==Type.DELIVERY && flow==Flow.TRANSFER; }
/** 按原批次顺序读取，仅对固定步骤附加耗时，不更改来源或失败语义。 */
public List<Fact> read(List<FactRequest> requests) {
var mapped=new HashMap<String,List<AppointmentWorkSourceMapper.Source>>();
var privateSources=new HashMap<String,List<AppointmentWorkSourceMapper.Source>>();
var containers=requests.stream().filter(r->r.batchKey()==null && r.requirementKey()!=null && r.requirementKey().startsWith("INBOUND:")).map(FactRequest::containerBookingId).distinct().toList();
if(!containers.isEmpty()) sourceMapping.containerSourcesBatch(containers).forEach(s->mapped.computeIfAbsent(s.containerBookingId(),ignored->new ArrayList<>()).add(s.source()));
if(!containers.isEmpty()) measure(Stage.DELIVERY_SOURCE_MAP,()->sourceMapping.containerSourcesBatch(containers)).forEach(s->mapped.computeIfAbsent(s.containerBookingId(),ignored->new ArrayList<>()).add(s.source()));
for(var request:requests) if(request.batchKey()==null && request.requirementKey()!=null && request.requirementKey().startsWith("INBOUND:")) {
var sources=mapped.getOrDefault(request.containerBookingId(),List.of());
if(sources.size()>1000) throw unavailable("柜次尾程来源超过完整读取上限");
@@ -40,20 +42,20 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
var refs=requests.stream().filter(r->r.batchKey()==null && !privateSources.containsKey(r.workItemKey())).flatMap(r->r.sources().stream())
.map(s->new AppointmentCoverageService.SourceRef(s.customerId(),s.sourceKey(),s.destinationKey())).distinct().toList();
var inspections=new HashMap<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection>();
if(!refs.isEmpty()) cargo.inspectBatch(refs).forEach(i->inspections.put(i.reference(),i));
if(!refs.isEmpty()) measure(Stage.DELIVERY_CARGO,()->cargo.inspectBatch(refs)).forEach(i->inspections.put(i.reference(),i));
var batches=requests.stream().map(FactRequest::batchKey).filter(k->k!=null && !k.startsWith("DISPATCH:")).distinct().toList();
var snapshots=new HashMap<String,FbaAppointmentResponse>();
var deliveryRows=new HashMap<String,List<AppointmentWorkDeliveryMapper.Delivery>>();
Map<String,String> impacts=Map.of();
var receivingSnapshots=new HashMap<String,BatchReceiving>();
if(!batches.isEmpty()) {
appointments.batchSnapshots(batches).forEach(a->snapshots.put(a.batchKey(),a));
deliveries.batches(batches).forEach(d->deliveryRows.computeIfAbsent(d.batchKey(),ignored->new ArrayList<>()).add(d.delivery()));
receiving.snapshots(batches).forEach(s->receivingSnapshots.put(s.batchKey(),s));
impacts=commitments.impactTokens(requests.stream().filter(r->r.batchKey()!=null && r.batchKey().startsWith("FBA_APPOINTMENT:")).map(r->customerRequirement(r.batchKey(),deliveryCustomer(r.sources()))).distinct().toList());
measure(Stage.DELIVERY_SNAPSHOTS,()->appointments.batchSnapshots(batches)).forEach(a->snapshots.put(a.batchKey(),a));
measure(Stage.DELIVERY_ROWS,()->deliveries.batches(batches)).forEach(d->deliveryRows.computeIfAbsent(d.batchKey(),ignored->new ArrayList<>()).add(d.delivery()));
measure(Stage.DELIVERY_RECEIVING,()->receiving.snapshots(batches)).forEach(s->receivingSnapshots.put(s.batchKey(),s));
impacts=measure(Stage.DELIVERY_IMPACTS,()->commitments.impactTokens(requests.stream().filter(r->r.batchKey()!=null && r.batchKey().startsWith("FBA_APPOINTMENT:")).map(r->customerRequirement(r.batchKey(),deliveryCustomer(r.sources()))).distinct().toList()));
}
var context=new BatchFacts(snapshots,deliveryRows,impacts,receivingSnapshots);
return requests.stream().map(r->read(r,privateSources.get(r.workItemKey()),inspections,context)).toList();
return measure(Stage.DELIVERY_ASSEMBLY,()->requests.stream().map(r->read(r,privateSources.get(r.workItemKey()),inspections,context)).toList());
}
private record BatchFacts(Map<String,FbaAppointmentResponse> appointments,Map<String,List<AppointmentWorkDeliveryMapper.Delivery>> deliveries,Map<String,String> impacts,Map<String,BatchReceiving> receiving) { }
private Fact read(FactRequest request,List<AppointmentWorkSourceMapper.Source> privateSources,Map<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection> inspections,BatchFacts context) {
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/work/AppointmentWorkRequirementProvider.java
+27 -75
View File
@@ -17,85 +17,29 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
private final AppointmentCargoVerificationService cargo;
private final InboundFbaAppointmentService appointments;
private final AppointmentWorkSourceMapper sourceMapping;
@jakarta.annotation.Resource private AppointmentRequirementReadRepository reads;
public AppointmentWorkRequirementProvider(AppointmentSourceService sources,AppointmentCargoVerificationService cargo,InboundFbaAppointmentService appointments,AppointmentWorkSourceMapper sourceMapping) {
this.sources=sources; this.cargo=cargo; this.appointments=appointments; this.sourceMapping=sourceMapping;
}
/** 本次生成共享原料，实例不会跨同步页或独立命令保存。 */
public RequirementReader openRequirements(List<RequirementInput> inputs) {
/** 同次生成仅共享按需读取的目录；原业务数据继续逐需求核实。 */
public RequirementReader openRequirements() {
return new RequirementReader() {
private Prepared prepared;
/** 原料提示不替代实际路线判断；未提示的真实需求按同一实现即时读取。 */
private AppointmentSourceService.ReadScope scope;
/** 直送和未建立尾程来源的需求不创建目录读取依赖。 */
public List<Requirement> read(ContainerSource container,Flow flow) {
if(flow==Flow.DIRECT) return original(container);
if(flow!=Flow.TRANSFER) throw conflict("需求清单必须明确有效作业路线");
if(prepared==null) {
var requested=new ArrayList<>(inputs);requested.add(new RequirementInput(container,flow));
prepared=prepare(requested);
}
var data=prepared.containers.contains(container)?prepared:prepare(List.of(new RequirementInput(container,flow)));
return requirements(container,flow,data);
return requirements(container,flow,refs->{
if(scope==null) scope=sources.openReadScope();
return scope.resolveBatch(refs);
});
}
};
}
/** 私人地址仍以原客户仓需求表示，直送不增加尾程读取依赖。 */
private List<Requirement> original(ContainerSource container) {
return container.requirements().stream().map(r->"PRIVATE".equals(r.destinationKind())?new Requirement(r.requirementKey(),r.batchKey(),r.batchNo(),"CUSTOMER",r.originName(),r.destinationName(),r.timezone(),r.references(),r.replacesRequirementKeys(),r.sources()):r).toList();
}
private record Group(ContainerSource container,String requirementKey) { }
/** 只含本次原料及延后执行器，不缓存已经解析的业务响应。 */
private static final class Prepared {
private final Set<ContainerSource> containers=new LinkedHashSet<>();
private final Map<String,List<AppointmentWorkSourceMapper.Source>> mapped=new HashMap<>();
private final Map<Group,Integer> ordinals=new LinkedHashMap<>();
private List<AppointmentSourceService.OrderGroup> sources=List.of();
private RuntimeException mappingError,sourceError;
private java.util.function.IntFunction<List<FbaAppointmentResponse>> history;
private java.util.function.Function<AppointmentSourceService.Source,AppointmentCargoVerificationService.Context> inspection;
}
/** 先批量读取原料，组级业务错误在原需求访问位置抛出。 */
private Prepared prepare(List<RequirementInput> inputs) {
var data=new Prepared();
inputs.stream().filter(i->i.flow()==Flow.TRANSFER).map(RequirementInput::source).forEach(data.containers::add);
var ids=data.containers.stream().map(ContainerSource::containerBookingId).distinct().toList();
try {
for(int start=0;start<ids.size();start+=100) reads.sources(ids.subList(start,Math.min(start+100,ids.size())))
.forEach(row->data.mapped.computeIfAbsent(row.containerBookingId(),ignored->new ArrayList<>()).add(row.source()));
} catch(RuntimeException error) { data.mappingError=error;return data; }
var groups=new ArrayList<List<AppointmentExternalSourceGateway.OrderRef>>();
for(var container:data.containers) {
var mapped=data.mapped.getOrDefault(container.containerBookingId(),List.of());
if(mapped.size()>1000) continue;
for(var requirement:original(container)) {
if(!"FBA".equals(requirement.destinationKind())) continue;
var inbound=requirement.requirementKey().startsWith("INBOUND:")?requirement.requirementKey().substring(8):null;
var orders=mapped.stream().filter(row->Objects.equals(row.sourceInboundId(),inbound)).toList();
if(orders.isEmpty()) continue;
data.ordinals.put(new Group(container,requirement.requirementKey()),groups.size());
groups.add(orders.stream().map(row->new AppointmentExternalSourceGateway.OrderRef(row.customerId(),row.id())).toList());
}
}
if(groups.isEmpty()) return data;
try { data.sources=sources.openReadScope().resolveOrderGroups(groups); }
catch(RuntimeException error) { data.sourceError=error;return data; }
var historyGroups=data.sources.stream().map(group->group.error()==null?group.sources().stream().map(source->new AppointmentCoverageService.SourceRef(source.customerId(),source.sourceKey(),source.destinationKey())).toList():List.<AppointmentCoverageService.SourceRef>of()).toList();
var valid=data.sources.stream().filter(group->group.error()==null).flatMap(group->group.sources().stream()).toList();
if(valid.isEmpty()) return data;
try { data.history=appointments.prepareRelatedBatchSnapshots(historyGroups); }
catch(RuntimeException error) { data.history=ignored->{throw error;}; }
// 不提前解析货件，仍由原 history -> anyMatch 路径按实际访问顺序执行。
try { data.inspection=cargo.prepareInspection(valid); }
catch(RuntimeException error) { data.inspection=ignored->{throw error;}; }
return data;
}
/** 按当前实物分配生成客户需求，保留已撤销批次身份以退役旧事项。 */
private List<Requirement> requirements(ContainerSource container,Flow effectiveFlow,
Prepared data) {
var original=original(container);
java.util.function.Function<List<AppointmentCoverageService.SourceRef>,List<AppointmentSourceService.Source>> resolver) {
var original=container.requirements().stream().map(r->"PRIVATE".equals(r.destinationKind())?new Requirement(r.requirementKey(),r.batchKey(),r.batchNo(),"CUSTOMER",r.originName(),r.destinationName(),r.timezone(),r.references(),r.replacesRequirementKeys(),r.sources()):r).toList();
if(effectiveFlow==Flow.DIRECT) return original;
if(effectiveFlow!=Flow.TRANSFER) throw conflict("需求清单必须明确有效作业路线");
if(data.mappingError!=null) throw data.mappingError;
var mapped=data.mapped.getOrDefault(container.containerBookingId(),List.of());
var mapped=sourceMapping.containerSources(container.containerBookingId());
if(mapped.size()>1000) throw unavailable("柜次尾程来源超过完整读取上限");
var result=new LinkedHashMap<String,Requirement>();
var privateDispatches=original.stream().anyMatch(r->"CUSTOMER".equals(r.destinationKind()))?sourceMapping.privateDispatches(container.containerBookingId()):List.<AppointmentWorkSourceMapper.DispatchSource>of();
@@ -106,13 +50,9 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
if(sourceOrders.isEmpty()) { result.put(requirement.requirementKey(),requirement); continue; }
if("CUSTOMER".equals(requirement.destinationKind())) { customer(requirement,sourceOrders,privateDispatches,result);continue; }
if(!"FBA".equals(requirement.destinationKind())) { result.put(requirement.requirementKey(),requirement); continue; }
if(sourceOrders.size()>100) throw unavailable("Firstleg需求缺少完整客户货源关系");
if(data.sourceError!=null) throw data.sourceError;
int ordinal=data.ordinals.get(new Group(container,requirement.requirementKey()));
var group=data.sources.get(ordinal);
var refs=group.references();
var refs=canonical(sourceOrders.stream().map(s->new SourceReference(s.customerId(),"ORDER:"+s.id(),null)).toList());
List<AppointmentSourceService.Source> resolved;
try { resolved=group.requireSources(); }
try { resolved=resolver.apply(refs); }
catch(org.springframework.web.server.ResponseStatusException error) {
if(!AppointmentWorkSourceAvailability.unresolved(error)) throw error;
// Keep the complete physical obligation and its canonical source identities. The fact
@@ -123,10 +63,10 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
continue;
}
var complete=resolved.stream().map(s->new AppointmentCoverageService.SourceRef(s.customerId(),s.sourceKey(),s.destinationKey())).toList();
var history=data.history.apply(ordinal);
var history=appointments.relatedBatchSnapshots(complete);
var active=history.stream().filter(AppointmentWorkRequirementProvider::active).toList();
// Unknown remaining quantity is still work, never a zero or an absent requirement.
boolean remainder=resolved.stream().map(data.inspection).anyMatch(c->c.availablePackages()==null || c.availablePackages()>0);
boolean remainder=resolved.stream().map(cargo::inspect).anyMatch(c->c.availablePackages()==null || c.availablePackages()>0);
var replaced=history.stream().filter(a->!active(a)).map(FbaAppointmentResponse::batchKey).toList();
if(remainder) {
var restored=new LinkedHashSet<>(replaced);
@@ -188,6 +128,18 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
result.put(batch,new Requirement(batch,batch,all.get(0).dispatchNo(),"CUSTOMER",requirement.originName(),requirement.destinationName(),requirement.timezone(),refs.stream().distinct().toList(),List.copyOf(replacements),all.stream().map(d->new SourceReference(d.customerId(),d.sourceKey(),d.destinationKey())).distinct().toList()));
}
}
private List<AppointmentCoverageService.SourceRef> canonical(List<SourceReference> input) {
if(input==null || input.isEmpty() || input.size()>100) throw unavailable("Firstleg需求缺少完整客户货源关系");
var orders=input.stream().filter(s->s.sourceKey()!=null && s.sourceKey().startsWith("ORDER:")).map(s->{
try { return new AppointmentExternalSourceGateway.OrderRef(s.customerId(),Long.parseLong(s.sourceKey().substring(6))); }
catch(RuntimeException error) { throw unavailable("Firstleg原订单关联无效",error); }
}).toList();
var result=new ArrayList<AppointmentCoverageService.SourceRef>();
if(!orders.isEmpty()) result.addAll(sources.canonicalOrders(orders));
input.stream().filter(s->s.sourceKey()!=null && s.sourceKey().startsWith("BOOKING:")).map(s->new AppointmentCoverageService.SourceRef(s.customerId(),s.sourceKey(),s.destinationKey())).forEach(result::add);
if(result.size()!=input.size()) throw unavailable("来源键必须明确为ORDER或BOOKING");
return result.stream().distinct().toList();
}
/** 已取消且没有当前实物分配的记录只属历史；取消但仍有货物的批次继续产生需求。 */
private static boolean active(FbaAppointmentResponse batch) {
if(batch.replacementBatchKeys()!=null && !batch.replacementBatchKeys().isEmpty()) return false;