關(guān)門信號(hào) → 啟動(dòng)異步任務(wù) → 數(shù)據(jù)收集 → 三重校驗(yàn) → 支付扣款 → 狀態(tài)更新

// 訂單實(shí)體
@Entity
@Table(name = "vending_orders")
@Data
public class VendingOrder {
    @Id
    private String orderId;
    private String userId;          // 用戶ID
    private String deviceId;        // 設(shè)備ID
    private String status;          // 狀態(tài): OPEN/CLOSED/PAID/FAILED
    private BigDecimal amount;      // 訂單金額
    private LocalDateTime createTime;
    private LocalDateTime updateTime;
}
// Redis緩存中的設(shè)備會(huì)話
@Data
public class DeviceSession {
    private String sessionId;
    private String deviceId;
    private String orderId;
    private String userId;
    private String status;          // 會(huì)話狀態(tài)
    private LocalDateTime startTime;
    private List<DeviceEvent> events; // 購物事件記錄
}
// 設(shè)備事件
@Data
public class DeviceEvent {
    private String eventId;
    private String deviceId;
    private String orderId;
    private String type;            // PICK/PUT_BACK
    private String productId;
    private LocalDateTime eventTime;
    private String position;        // 貨道位置
}

@RestController
@RequestMapping("/api/vending")
@Slf4j
public class VendingController {
    @Autowired
    private VendingService vendingService;
    @Autowired
    private SettlementService settlementService;
    /**
     * 掃碼開門接口
     */
    @PostMapping("/open")
    public ResponseEntity<ApiResponse> openDevice(
            @RequestParam String deviceId,
            @RequestParam String authToken) {
        log.info("收到開門請(qǐng)求: deviceId={}", deviceId);
        try {
            OpenResult result = vendingService.processOpenDevice(deviceId, authToken);
            return ResponseEntity.ok(ApiResponse.success(result));
        } catch (BusinessException e) {
            log.warn("開門業(yè)務(wù)異常: {}", e.getMessage());
            return ResponseEntity.badRequest().body(ApiResponse.error(e.getMessage()));
        }
    }
    /**
     * 查詢訂單狀態(tài)
     */
    @GetMapping("/order/{orderId}")
    public ResponseEntity<ApiResponse> getOrderStatus(@PathVariable String orderId) {
        VendingOrder order = vendingService.getOrderById(orderId);
        return ResponseEntity.ok(ApiResponse.success(order));
    }
}
// 統(tǒng)一API響應(yīng)格式
@Data
class ApiResponse {
    private boolean success;
    private String message;
    private Object data;
    public static ApiResponse success(Object data) {
        ApiResponse response = new ApiResponse();
        response.setSuccess(true);
        response.setData(data);
        return response;
    }
    public static ApiResponse error(String message) {
        ApiResponse response = new ApiResponse();
        response.setSuccess(false);
        response.setMessage(message);
        return response;
    }
}

@Service
@Slf4j
public class VendingService {
    @Autowired
    private UserAuthService authService;
    @Autowired
    private DeviceService deviceService;
    @Autowired
    private OrderService orderService;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private MqttService mqttService;
    /**
     * 處理開門請(qǐng)求的核心邏輯
     */
    public OpenResult processOpenDevice(String deviceId, String authToken) {
        // 1. 用戶身份驗(yàn)證
        String userId = authService.verifyWechatToken(authToken);
        if (userId == null) {
            throw new BusinessException("用戶身份驗(yàn)證失敗");
        }
        // 2. 檢查設(shè)備狀態(tài)
        DeviceStatus deviceStatus = deviceService.getDeviceStatus(deviceId);
        if (!deviceStatus.isAvailable()) {
            throw new BusinessException("設(shè)備暫不可用: " + deviceStatus.getStatus());
        }
        // 3. 創(chuàng)建訂單
        VendingOrder order = orderService.createOrder(userId, deviceId);
        log.info("創(chuàng)建訂單成功: orderId={}, userId={}, deviceId={}", 
                order.getOrderId(), userId, deviceId);
        // 4. 創(chuàng)建設(shè)備會(huì)話并緩存
        DeviceSession session = createDeviceSession(deviceId, order);
        cacheDeviceSession(session);
        // 5. 鎖定設(shè)備，避免重復(fù)開門
        deviceService.lockDevice(deviceId, order.getOrderId());
        // 6. 發(fā)送開門指令
        mqttService.sendOpenCommand(deviceId);
        // 7. 請(qǐng)求設(shè)備上報(bào)初始狀態(tài)
        mqttService.requestInitialSnapshot(deviceId);
        return new OpenResult(order.getOrderId(), deviceId, "開門指令已發(fā)送");
    }
    private DeviceSession createDeviceSession(String deviceId, VendingOrder order) {
        DeviceSession session = new DeviceSession();
        session.setSessionId(UUID.randomUUID().toString());
        session.setDeviceId(deviceId);
        session.setOrderId(order.getOrderId());
        session.setUserId(order.getUserId());
        session.setStatus("OPEN");
        session.setStartTime(LocalDateTime.now());
        session.setEvents(new ArrayList<>());
        return session;
    }
    private void cacheDeviceSession(DeviceSession session) {
        String key = buildSessionKey(session.getDeviceId());
        redisTemplate.opsForValue().set(key, session, Duration.ofMinutes(10));
        log.debug("設(shè)備會(huì)話已緩存: key={}", key);
    }
    private String buildSessionKey(String deviceId) {
        return "vending:session:" + deviceId;
    }
}

@Component
@Slf4j
public class MqttMessageHandler {
    @Autowired
    private VendingService vendingService;
    @Autowired
    private SettlementService settlementService;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 處理關(guān)門事件 - 觸發(fā)結(jié)算流程
     */
    @MqttListener(topics = "device/${spring.application.env}/+/event/door_close")
    public void handleDoorClose(String message) {
        try {
            DoorCloseEvent event = JSON.parseObject(message, DoorCloseEvent.class);
            log.info("收到關(guān)門事件: deviceId={}", event.getDeviceId());
            // 異步處理結(jié)算，不阻塞MQTT線程
            CompletableFuture.runAsync(() -> {
                settlementService.startSettlementProcess(event.getDeviceId());
            });
        } catch (Exception e) {
            log.error("處理關(guān)門事件失敗: message={}", message, e);
        }
    }
    /**
     * 處理商品拿取/放回事件
     */
    @MqttListener(topics = "device/${spring.application.env}/+/event/product")
    public void handleProductEvent(String message) {
        try {
            ProductEvent event = JSON.parseObject(message, ProductEvent.class);
            log.debug("處理商品事件: deviceId={}, type={}, product={}", 
                     event.getDeviceId(), event.getEventType(), event.getProductId());
            // 記錄到Redis緩存
            recordProductEvent(event);
        } catch (Exception e) {
            log.error("處理商品事件失敗: message={}", message, e);
        }
    }
    /**
     * 處理設(shè)備上報(bào)的初始/最終快照
     */
    @MqttListener(topics = "device/${spring.application.env}/+/snapshot")
    public void handleSnapshot(String message) {
        try {
            DeviceSnapshot snapshot = JSON.parseObject(message, DeviceSnapshot.class);
            log.info("處理設(shè)備快照: deviceId={}, type={}", 
                    snapshot.getDeviceId(), snapshot.getSnapshotType());
            // 存儲(chǔ)快照數(shù)據(jù)，用于后續(xù)對(duì)比分析
            storeDeviceSnapshot(snapshot);
        } catch (Exception e) {
            log.error("處理快照失敗: message={}", message, e);
        }
    }
    private void recordProductEvent(ProductEvent event) {
        String sessionKey = "vending:session:" + event.getDeviceId();
        DeviceSession session = (DeviceSession) redisTemplate.opsForValue().get(sessionKey);
        if (session != null) {
            DeviceEvent deviceEvent = convertToDeviceEvent(event, session.getOrderId());
            session.getEvents().add(deviceEvent);
            redisTemplate.opsForValue().set(sessionKey, session, Duration.ofMinutes(10));
        }
    }
}

@Service
@Slf4j
public class SettlementService {
    @Autowired
    private OrderService orderService;
    @Autowired
    private DeviceService deviceService;
    @Autowired
    private PaymentService paymentService;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private NotificationService notificationService;
    /**
     * 啟動(dòng)結(jié)算流程
     */
    @Async("settlementExecutor")
    public void startSettlementProcess(String deviceId) {
        log.info("開始結(jié)算流程: deviceId={}", deviceId);
        try {
            // 1. 獲取設(shè)備會(huì)話
            DeviceSession session = getDeviceSession(deviceId);
            if (session == null) {
                log.error("設(shè)備會(huì)話不存在: deviceId={}", deviceId);
                return;
            }
            // 2. 更新訂單狀態(tài)為結(jié)算中
            orderService.updateOrderStatus(session.getOrderId(), "SETTLING");
            // 3. 獲取設(shè)備上報(bào)的最終數(shù)據(jù)
            SettlementData settlementData = collectSettlementData(deviceId);
            // 4. 執(zhí)行結(jié)算計(jì)算
            SettlementResult result = calculateSettlement(settlementData);
            // 5. 處理支付
            boolean paymentSuccess = processPayment(session, result);
            // 6. 更新訂單狀態(tài)
            updateOrderAfterSettlement(session, result, paymentSuccess);
            // 7. 清理資源
            cleanupAfterSettlement(deviceId, session.getOrderId());
            log.info("結(jié)算流程完成: deviceId={}, orderId={}, success={}", 
                    deviceId, session.getOrderId(), paymentSuccess);
        } catch (Exception e) {
            log.error("結(jié)算流程異常: deviceId={}", deviceId, e);
            handleSettlementFailure(deviceId, e);
        }
    }
    /**
     * 收集結(jié)算所需的所有數(shù)據(jù)
     */
    private SettlementData collectSettlementData(String deviceId) {
        SettlementData data = new SettlementData();
        // 獲取初始和最終快照
        data.setInitialSnapshot(deviceService.getInitialSnapshot(deviceId));
        data.setFinalSnapshot(deviceService.getFinalSnapshot(deviceId));
        // 獲取重量傳感器數(shù)據(jù)
        data.setWeightData(deviceService.getWeightSensorData(deviceId));
        // 獲取購物事件記錄
        data.setProductEvents(getRecordedEvents(deviceId));
        return data;
    }
    /**
     * 核心結(jié)算算法 - 三重校驗(yàn)
     */
    private SettlementResult calculateSettlement(SettlementData data) {
        // 1. 視覺對(duì)比分析
        List<Product> visualProducts = analyzeVisualChanges(
            data.getInitialSnapshot(), 
            data.getFinalSnapshot()
        );
        // 2. 重量變化分析
        List<Product> weightProducts = analyzeWeightChanges(data.getWeightData());
        // 3. 事件記錄分析
        List<Product> eventProducts = analyzeEventSequence(data.getProductEvents());
        // 4. 沖突解決和結(jié)果融合
        return resolveProductConflicts(visualProducts, weightProducts, eventProducts);
    }
    /**
     * 沖突解決策略
     */
    private SettlementResult resolveProductConflicts(List<Product> visualProducts, 
                                                    List<Product> weightProducts,
                                                    List<Product> eventProducts) {
        SettlementResult result = new SettlementResult();
        // 策略1: 視覺識(shí)別優(yōu)先（最直接證據(jù)）
        Map<String, Product> productMap = new HashMap<>();
        // 首先信任視覺識(shí)別結(jié)果
        for (Product product : visualProducts) {
            productMap.put(product.getPosition(), product);
        }
        // 用重量數(shù)據(jù)驗(yàn)證和補(bǔ)充
        for (Product weightProduct : weightProducts) {
            Product visualProduct = productMap.get(weightProduct.getPosition());
            if (visualProduct == null) {
                // 視覺沒識(shí)別到但重量有變化，信任重量數(shù)據(jù)
                productMap.put(weightProduct.getPosition(), weightProduct);
            }
        }
        // 用事件記錄進(jìn)行最終校驗(yàn)
        result.setFinalProducts(new ArrayList<>(productMap.values()));
        result.setConflictResolved(true);
        log.debug("沖突解決完成: 視覺識(shí)別{}個(gè), 重量變化{}個(gè), 最終確認(rèn){}個(gè)", 
                 visualProducts.size(), weightProducts.size(), result.getFinalProducts().size());
        return result;
    }
}

@Service
@Slf4j
public class PaymentService {
    @Autowired
    private WechatPayService wechatPayService;
    @Autowired
    private OrderService orderService;
    /**
     * 執(zhí)行支付
     */
    public boolean processPayment(DeviceSession session, SettlementResult result) {
        try {
            PaymentRequest request = new PaymentRequest();
            request.setUserId(session.getUserId());
            request.setOrderId(session.getOrderId());
            request.setAmount(calculateTotalAmount(result.getFinalProducts()));
            request.setDescription("智能售貨柜購物");
            PaymentResponse response = wechatPayService.unifiedOrder(request);
            if ("SUCCESS".equals(response.getResultCode())) {
                log.info("支付成功: orderId={}, amount={}", 
                        session.getOrderId(), request.getAmount());
                return true;
            } else {
                log.warn("支付失敗: orderId={}, error={}", 
                        session.getOrderId(), response.getErrMsg());
                return false;
            }
        } catch (Exception e) {
            log.error("支付處理異常: orderId={}", session.getOrderId(), e);
            return false;
        }
    }
    private BigDecimal calculateTotalAmount(List<Product> products) {
        return products.stream()
                .map(Product::getPrice)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
}

@Configuration
@EnableAsync
@EnableScheduling
public class AsyncConfig {
    @Bean("settlementExecutor")
    public TaskExecutor settlementTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("settlement-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}
@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        return template;
    }
}