perf(lastleg): batch receiving candidates and trace slow work invalidation

This commit is contained in:
2026-10-17 07:49:56 +08:00
parent af97cd17d0
commit 5fd74e9c8c
9 changed files with 377 additions and 17 deletions
@@ -4,13 +4,15 @@ import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/** Original business commands invalidate the manifest atomically with their business write. */
/** 原业务写入与工作清单失效保持在同一事务,不延后或绕过版本门禁。 */
@Component
public class AppointmentWorkChanges {
private final AppointmentWorkMapper mapper;
public AppointmentWorkChanges(AppointmentWorkMapper mapper) { this.mapper=mapper; }
private final AppointmentWorkStore store;
/** 复用工作事项持久层,统一记录守卫更新耗时。 */
public AppointmentWorkChanges(AppointmentWorkStore store) { this.store=store; }
/** 只标记需求变化,保留调用方原事务和异常语义。 */
@Transactional(propagation=Propagation.MANDATORY)
public void requirementsChanged() {
if(mapper.invalidateRequirements()!=1) throw AppointmentWorkPolicy.unavailable("工作事项同步守卫未初始化");
store.invalidateRequirements();
}
}
@@ -1,6 +1,9 @@
package com.baiyue.lastleg.appointmentwork;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;
import java.time.*;
import java.util.*;
@@ -9,10 +12,25 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@Component
public class AppointmentWorkStore {
private static final Logger LOG = LoggerFactory.getLogger(AppointmentWorkStore.class);
public static final int MAX_ITEMS=10000;
private final AppointmentWorkMapper mapper;
private final ObjectMapper json;
public AppointmentWorkStore(AppointmentWorkMapper mapper,ObjectMapper json) { this.mapper=mapper; this.json=json; }
/** 只测量守卫更新语句,不改变锁、事务或版本,用于区分慢写是否卡在该语句。 */
public void invalidateRequirements() {
long started = System.nanoTime();
try {
if (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"));
}
}
}
public AppointmentWorkMapper.Guard lock() {
var guard=mapper.lockGuard(); if(guard==null) throw unavailable("工作事项DDL尚未集成"); return guard;
}
@@ -0,0 +1,35 @@
package com.baiyue.lastleg.inboundplan.pda.mapper;
import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import java.util.List;
/** 候选列表的批量资源门禁与实收汇总,不改变原收货命令。 */
@Mapper
public interface ReceivingCandidateMapper {
/** 柜次和柜号由数据库按原门禁的比较语义匹配,避免 Java 大小写比较缩小候选。 */
@Select("SELECT o.id AS order_id,o.source_container_booking_id AS booking_id FROM ll_order o " +
"JOIN ll_container_shipping_gate g ON g.container_booking_id=o.source_container_booking_id " +
"AND g.container_no=o.container_no WHERE g.status='RECEIVING_AVAILABLE'")
List<AvailableOrder> availableOrders();
record AvailableOrder(Long orderId, String bookingId) { }
/** 保留原转换命令的客户和计划精确匹配,一批读取未确认计划。 */
@Select("<script>SELECT p.id FROM ll_inbound_plan p JOIN ll_order o ON o.customer_id=p.customer_id " +
"AND o.id=p.source_order_id WHERE p.id IN " +
"<foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach> AND " +
ContainerPlanConversionCommandMapper.UNCONFIRMED_PLAN + "</script>")
List<Long> unconfirmedPlans(@Param("ids") List<Long> ids);
/** 实收始终按客户、计划及计划行三重身份汇总。 */
@Select("<script>SELECT customer_id,plan_id,line_id,COALESCE(SUM(picked_package_count),0) AS received " +
"FROM ll_inbound_plan_pallet_allocation WHERE line_id IN " +
"<foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach> " +
"GROUP BY customer_id,plan_id,line_id</script>")
List<Received> received(@Param("ids") List<Long> ids);
record Received(String customerId, Long planId, Long lineId, int received) { }
}
@@ -0,0 +1,76 @@
package com.baiyue.lastleg.inboundplan.pda.repository;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baiyue.lastleg.entity.LastlegOrder;
import com.baiyue.lastleg.mapper.LastlegOrderMapper;
import com.baiyue.lastleg.inboundplan.entity.InboundPlan;
import com.baiyue.lastleg.inboundplan.entity.InboundPlanLine;
import com.baiyue.lastleg.inboundplan.mapper.InboundPlanMapper;
import com.baiyue.lastleg.inboundplan.mapper.InboundPlanLineMapper;
import com.baiyue.lastleg.inboundplan.pda.mapper.ReceivingCandidateMapper;
import com.baiyue.lastleg.inboundplan.pda.mapper.ReceivingContainerSessionMapper;
import com.baiyue.lastleg.inboundplan.pda.entity.ReceivingContainerSession;
import org.springframework.stereotype.Repository;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
/** 一次请求内批量读取候选所需事实,避免按柜、计划和货件逐条查询。 */
@Repository
public class ReceivingCandidateRepository {
private static final int CHUNK_SIZE = 500;
private final LastlegOrderMapper orders;
private final InboundPlanMapper plans;
private final InboundPlanLineMapper lines;
private final ReceivingCandidateMapper candidates;
private final ReceivingContainerSessionMapper sessions;
/** 仅依赖当前收货业务的本库 Mapper。 */
public ReceivingCandidateRepository(LastlegOrderMapper orders, InboundPlanMapper plans,
InboundPlanLineMapper lines, ReceivingCandidateMapper candidates, ReceivingContainerSessionMapper sessions) {
this.orders=orders; this.plans=plans; this.lines=lines; this.candidates=candidates; this.sessions=sessions;
}
/** 先读取当前账号会话,供服务在加载候选前核对仓库归属。 */
public ReceivingContainerSession currentSession(String activeOperatorKey) {
return sessions.findActiveByOperator(activeOperatorKey);
}
/** 读取柜号占用事实,由服务排除其他操作账号正在处理的柜。 */
public List<ReceivingContainerSession> activeSessions() {
return sessions.findActiveContainers();
}
public record LineKey(String customerId, Long planId, Long lineId) { }
public record Facts(List<LastlegOrder> orders, List<InboundPlan> plans, List<InboundPlanLine> lines,
Set<Long> availableOrderIds, Set<Long> unconfirmedPlans, Map<LineKey,Integer> received) { }
/** 先缩小到可收货柜,保留其全部客户订单,不能按当前仓提前裁掉柜内冲突记录。 */
public Facts load() {
var available=candidates.availableOrders();
var bookingIds=available.stream().map(ReceivingCandidateMapper.AvailableOrder::bookingId).distinct().toList();
var orderRows=batch(bookingIds, ids->orders.selectList(new LambdaQueryWrapper<LastlegOrder>()
.in(LastlegOrder::getSourceContainerBookingId,ids).isNotNull(LastlegOrder::getContainerNo)))
.stream().collect(Collectors.toMap(LastlegOrder::getId,Function.identity(),(left,right)->left))
.values().stream().sorted(Comparator.comparing(LastlegOrder::getSourceContainerBookingId)
.thenComparing(LastlegOrder::getId)).toList();
var lineRows=batch(orderRows.stream().map(LastlegOrder::getId).toList(), ids->lines.selectList(
new LambdaQueryWrapper<InboundPlanLine>().in(InboundPlanLine::getOrderId,ids)))
.stream().sorted(Comparator.comparing(InboundPlanLine::getId)).toList();
var planIds=lineRows.stream().map(InboundPlanLine::getPlanId).filter(Objects::nonNull).distinct().toList();
var planRows=batch(planIds, ids->plans.selectList(new LambdaQueryWrapper<InboundPlan>().in(InboundPlan::getId,ids)));
var unconfirmed=new HashSet<>(batch(planIds,candidates::unconfirmedPlans));
var received=batch(lineRows.stream().map(InboundPlanLine::getId).toList(),candidates::received).stream()
.collect(Collectors.toMap(r->new LineKey(r.customerId(),r.planId(),r.lineId()),ReceivingCandidateMapper.Received::received));
return new Facts(orderRows,planRows,lineRows,available.stream().map(ReceivingCandidateMapper.AvailableOrder::orderId)
.collect(Collectors.toSet()),unconfirmed,received);
}
/** 固定大小分块,空范围不执行 SQL,也不截断候选。 */
private static <I,O> List<O> batch(List<I> ids,Function<List<I>,List<O>> query) {
var result=new ArrayList<O>();
for(int start=0;start<ids.size();start+=CHUNK_SIZE)
result.addAll(query.apply(ids.subList(start,Math.min(ids.size(),start+CHUNK_SIZE))));
return result;
}
}
@@ -1,6 +1,7 @@
package com.baiyue.lastleg.inboundplan.pda.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baiyue.lastleg.inboundplan.pda.repository.ReceivingCandidateRepository;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baiyue.lastleg.entity.LastlegOrder;
import com.baiyue.lastleg.entity.LastlegOrderDetail;
@@ -82,6 +83,7 @@ public class ReceivingContainerService {
private static final String PALLET_CAPACITY_MESSAGE =
"一个托盘最多只能收66箱,请拆分到其他托盘";
@Resource private ReceivingCandidateRepository candidateRepository;
@Resource private InboundPlanMapper plans;
@Resource private InboundPlanLineMapper lines;
@Resource private InboundPlanPalletMapper pallets;
@@ -119,24 +121,31 @@ public class ReceivingContainerService {
Long.toString(value.warehouseId()), value.warehouseCode(), value.warehouseName());
}
/** 批量读取候选事实,柜内客户、仓库一致性和排序仍按原规则判断。 */
public List<ReceivingContainerResponses.Candidate> candidates(String operatorKey, long overseasWarehouseId) {
ReceivingContainerSession current = sessions.findActiveByOperator(activeOperatorKey(operatorKey));
ReceivingContainerSession current = candidateRepository.currentSession(activeOperatorKey(operatorKey));
requireSessionWarehouse(current, overseasWarehouseId);
List<LastlegOrder> orderFacts = orders.selectList(new LambdaQueryWrapper<LastlegOrder>()
.isNotNull(LastlegOrder::getSourceContainerBookingId)
.isNotNull(LastlegOrder::getContainerNo)
.orderByAsc(LastlegOrder::getSourceContainerBookingId, LastlegOrder::getId));
List<InboundPlan> planFacts = plans.selectList(new LambdaQueryWrapper<InboundPlan>().orderByAsc(InboundPlan::getId));
List<InboundPlanLine> lineFacts = lines.selectList(new LambdaQueryWrapper<InboundPlanLine>().orderByAsc(InboundPlanLine::getId));
var facts = candidateRepository.load();
List<LastlegOrder> orderFacts = facts.orders();
Map<Long, List<InboundPlanLine>> linesByOrder = facts.lines().stream()
.collect(Collectors.groupingBy(InboundPlanLine::getOrderId));
Map<Long, InboundPlan> plansById = facts.plans().stream()
.collect(Collectors.toMap(InboundPlan::getId, Function.identity()));
Map<String, List<LastlegOrder>> grouped = orderFacts.stream()
.filter(order -> trim(order.getSourceContainerBookingId()) != null && trim(order.getContainerNo()) != null)
.collect(Collectors.groupingBy(LastlegOrder::getSourceContainerBookingId, LinkedHashMap::new, Collectors.toList()));
String operatorLockKey = activeOperatorKey(operatorKey);
Set<String> reservedByOthers = sessions.findActiveContainers().stream()
Set<String> reservedByOthers = candidateRepository.activeSessions().stream()
.filter(session -> !operatorLockKey.equals(session.getActiveOperatorKey()))
.map(ReceivingContainerSession::getActiveContainerKey).filter(Objects::nonNull).collect(Collectors.toSet());
return grouped.entrySet().stream()
.map(entry -> readiness(entry.getKey(), entry.getValue(), planFacts, lineFacts, overseasWarehouseId))
.map(entry -> {
var localLines = entry.getValue().stream().flatMap(order ->
linesByOrder.getOrDefault(order.getId(), List.of()).stream()).toList();
var localPlans = localLines.stream().map(line -> plansById.get(line.getPlanId()))
.filter(Objects::nonNull).distinct().toList();
return readiness(entry.getKey(), entry.getValue(), localPlans, localLines, overseasWarehouseId, facts);
})
.filter(Objects::nonNull)
.filter(ContainerReadiness::shippingReady)
.filter(value -> !value.allPlansCompleted())
@@ -679,9 +688,17 @@ public class ReceivingContainerService {
}
}
/** 写命令和详情继续即时复验原资源与船节点。 */
private ContainerReadiness readiness(String sourceContainerBookingId, List<LastlegOrder> orderFacts,
List<InboundPlan> planFacts, List<InboundPlanLine> lineFacts,
long overseasWarehouseId) {
return readiness(sourceContainerBookingId, orderFacts, planFacts, lineFacts, overseasWarehouseId, null);
}
/** 候选使用请求内批量事实;其他入口保留即时校验,避免改变写命令并发门禁。 */
private ContainerReadiness readiness(String sourceContainerBookingId, List<LastlegOrder> orderFacts,
List<InboundPlan> planFacts, List<InboundPlanLine> lineFacts,
long overseasWarehouseId, ReceivingCandidateRepository.Facts facts) {
Set<Long> orderIds = orderFacts.stream().map(LastlegOrder::getId).collect(Collectors.toSet());
Map<Long, List<InboundPlanLine>> linesByOrder = lineFacts.stream()
.filter(line -> orderIds.contains(line.getOrderId()))
@@ -718,9 +735,12 @@ public class ReceivingContainerService {
boolean allPlansCompleted = !matchedPlans.isEmpty()
&& matchedPlans.stream().allMatch(plan -> List.of("COMPLETED", "PENDING_RECONCILIATION")
.contains(plan.getStatus()));
boolean shippingReady = shippingGates.receivingAvailable(sourceContainerBookingId, containerNo) == 1;
boolean shippingReady = facts == null
? shippingGates.receivingAvailable(sourceContainerBookingId, containerNo) == 1
: facts.availableOrderIds().contains(orderFacts.get(0).getId());
boolean resourcesConfirmed = matchedPlans.stream().noneMatch(plan ->
plans.countUnconfirmedResourcePlans(plan.getCustomerId(), plan.getId()) > 0);
facts == null ? plans.countUnconfirmedResourcePlans(plan.getCustomerId(), plan.getId()) > 0
: facts.unconfirmedPlans().contains(plan.getId()));
boolean operable = allPlansReady && resourcesConfirmed && shippingReady && warehouseSnapshotConsistent;
int pending = Math.max(0, totalOrders - plannedOrders);
String blockReason = operable ? null : (!warehouseSnapshotConsistent
@@ -740,7 +760,9 @@ public class ReceivingContainerService {
List<InboundRef> matchedRefs = lineFacts.stream().filter(line -> orderIds.contains(line.getOrderId()))
.map(line -> new InboundRef(plansById.get(line.getPlanId()), line, ordersById.get(line.getOrderId())))
.filter(ref -> ref.plan() != null && ref.order() != null).toList();
List<ReceivingContainerResponses.Inbound> inboundValues = matchedRefs.stream().map(this::inbound).toList();
List<ReceivingContainerResponses.Inbound> inboundValues = matchedRefs.stream().map(ref -> facts == null
? inbound(ref) : inbound(ref, facts.received().getOrDefault(new ReceivingCandidateRepository.LineKey(
ref.order().getCustomerId(), ref.plan().getId(), ref.line().getId()), 0))).toList();
List<String> destinationCodes = inboundValues.stream().map(this::destinationCode)
.filter(Objects::nonNull).distinct().sorted().toList();
Totals totals = new Totals(inboundValues.size(), destinationCodes,
@@ -1152,12 +1174,18 @@ public class ReceivingContainerService {
return sessions.findActiveByContainer(activeContainerKey(containerNo));
}
/** 单柜详情和写命令即时读取本货件实收量。 */
private ReceivingContainerResponses.Inbound inbound(InboundRef ref) {
int received = allocations.selectList(new LambdaQueryWrapper<InboundPlanPalletAllocation>()
.eq(InboundPlanPalletAllocation::getCustomerId, ref.order().getCustomerId())
.eq(InboundPlanPalletAllocation::getPlanId, ref.plan().getId())
.eq(InboundPlanPalletAllocation::getLineId, ref.line().getId())).stream()
.mapToInt(value -> value(value.getPickedPackageCount())).sum();
return inbound(ref, received);
}
/** 共享相同展示投影,候选可使用批量汇总的实收量。 */
private ReceivingContainerResponses.Inbound inbound(InboundRef ref, int received) {
int total = value(ref.line().getRequiredPackageCount());
return new ReceivingContainerResponses.Inbound(ref.line().getInboundCode(), destinationType(ref.line(), ref.order()),
ref.plan().getOverseasWarehouseId() == null ? ref.order().getOverseasWarehouseId()