扩展补充篇

// 驱动端口：定义应用核心对外暴露的用例
public interface PlaceOrderUseCase {
    OrderResult placeOrder(PlaceOrderCommand command);
}

// 驱动端口：查询订单
public interface QueryOrderUseCase {
    OrderDto findById(String orderId);
}

// 被驱动端口：定义应用核心依赖的外部能力
public interface OrderRepository {
    Order save(Order order);
    Order findById(String orderId);
}

// 被驱动端口：通知服务
public interface NotificationService {
    void sendOrderConfirmation(Order order);
}

// 驱动适配器：REST 控制器实现了驱动端口的调用
@RestController
@RequestMapping("/orders")
public class OrderController {
    private final PlaceOrderUseCase placeOrderUseCase;
    private final QueryOrderUseCase queryOrderUseCase;

    public OrderController(PlaceOrderUseCase placeOrderUseCase,
                           QueryOrderUseCase queryOrderUseCase) {
        this.placeOrderUseCase = placeOrderUseCase;
        this.queryOrderUseCase = queryOrderUseCase;
    }

    @PostMapping
    public ResponseEntity<OrderResult> placeOrder(
            @RequestBody PlaceOrderRequest request) {
        PlaceOrderCommand command = new PlaceOrderCommand(
            request.getCustomerId(), request.getItems());
        return ResponseEntity.ok(placeOrderUseCase.placeOrder(command));
    }
}

// 被驱动适配器：JPA 实现，实现被驱动端口接口
@Repository
public class JpaOrderRepository implements OrderRepository {
    private final SpringDataOrderRepository jpaRepo;

    public JpaOrderRepository(SpringDataOrderRepository jpaRepo) {
        this.jpaRepo = jpaRepo;
    }

    @Override
    public Order save(Order order) {
        OrderEntity entity = OrderMapper.toEntity(order);
        OrderEntity saved = jpaRepo.save(entity);
        return OrderMapper.toDomain(saved);
    }

    @Override
    public Order findById(String orderId) {
        return jpaRepo.findById(orderId)
            .map(OrderMapper::toDomain)
            .orElse(null);
    }
}

// 命令对象：承载写操作的输入数据
public class PlaceOrderCommand {
    private final String customerId;
    private final List<OrderItem> items;

    public PlaceOrderCommand(String customerId, List<OrderItem> items) {
        this.customerId = customerId;
        this.items = items;
    }

    // getters...
}

// 命令处理器：执行写操作，使用领域模型保证业务规则
@Service
public class PlaceOrderCommandHandler {
    private final OrderRepository orderRepository;
    private final DomainEventPublisher eventPublisher;

    public PlaceOrderCommandHandler(OrderRepository orderRepository,
                                     DomainEventPublisher eventPublisher) {
        this.orderRepository = orderRepository;
        this.eventPublisher = eventPublisher;
    }

    public OrderResult handle(PlaceOrderCommand command) {
        // 使用领域模型执行业务规则
        Order order = Order.create(command.getCustomerId(), command.getItems());
        orderRepository.save(order);

        // 发布领域事件，供查询侧同步
        eventPublisher.publish(new OrderPlacedEvent(
            order.getId(), order.getCustomerId(), order.getTotalAmount()));

        return new OrderResult(order.getId(), "PLACED");
    }
}

// 查询对象：承载读操作的输入条件
public class FindOrderQuery {
    private final String orderId;
    // getters...
}

// 查询处理器：执行读操作，使用读优化模型
@Service
public class FindOrderQueryHandler {
    private final OrderReadRepository readRepository;

    public FindOrderQueryHandler(OrderReadRepository readRepository) {
        this.readRepository = readRepository;
    }

    // 查询直接读取投影表，无需经过领域模型
    public OrderDto handle(FindOrderQuery query) {
        return readRepository.findOrderDetail(query.getOrderId());
    }
}

// 读专用的 DTO，针对展示需求优化
public class OrderDto {
    private String orderId;
    private String customerName;  // 冗余存储，避免 JOIN 查询
    private List<OrderItemDto> items;
    private BigDecimal totalAmount;
    private String status;
    // 无业务逻辑，纯粹的数据传输对象
}

// 领域事件基类
public abstract class DomainEvent {
    private final String aggregateId;
    private final LocalDateTime timestamp;

    protected DomainEvent(String aggregateId) {
        this.aggregateId = aggregateId;
        this.timestamp = LocalDateTime.now();
    }

    public String getAggregateId() { return aggregateId; }
    public LocalDateTime getTimestamp() { return timestamp; }
}

// 具体事件：订单创建
public class OrderCreatedEvent extends DomainEvent {
    private final String customerId;
    private final List<OrderItem> items;

    public OrderCreatedEvent(String orderId, String customerId,
                             List<OrderItem> items) {
        super(orderId);
        this.customerId = customerId;
        this.items = items;
    }
    // getters...
}

// 具体事件：商品添加
public class ItemAddedEvent extends DomainEvent {
    private final OrderItem item;

    public ItemAddedEvent(String orderId, OrderItem item) {
        super(orderId);
        this.item = item;
    }
    // getters...
}

// 具体事件：订单取消
public class OrderCancelledEvent extends DomainEvent {
    private final String reason;

    public OrderCancelledEvent(String orderId, String reason) {
        super(orderId);
        this.reason = reason;
    }
    // getters...
}

// 聚合根：订单，通过事件重建状态
public class Order {
    private String id;
    private String customerId;
    private List<OrderItem> items = new ArrayList<>();
    private String status;
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();

    // 从事件重建：重放所有历史事件
    public static Order fromEvents(List<DomainEvent> events) {
        Order order = new Order();
        events.forEach(order::apply);
        order.uncommittedEvents.clear(); // 重建的事件不算新事件
        return order;
    }

    // 业务操作：创建订单
    public static Order create(String customerId, List<OrderItem> items) {
        Order order = new Order();
        OrderCreatedEvent event = new OrderCreatedEvent(
            UUID.randomUUID().toString(), customerId, items);
        order.apply(event);
        order.uncommittedEvents.add(event);
        return order;
    }

    // 业务操作：添加商品
    public void addItem(OrderItem item) {
        if ("CANCELLED".equals(status)) {
            throw new IllegalStateException("已取消的订单不能添加商品");
        }
        ItemAddedEvent event = new ItemAddedEvent(id, item);
        apply(event);
        uncommittedEvents.add(event);
    }

    // 业务操作：取消订单
    public void cancel(String reason) {
        if ("CANCELLED".equals(status)) {
            throw new IllegalStateException("订单已取消");
        }
        OrderCancelledEvent event = new OrderCancelledEvent(id, reason);
        apply(event);
        uncommittedEvents.add(event);
    }

    // 应用事件：修改聚合状态
    private void apply(DomainEvent event) {
        if (event instanceof OrderCreatedEvent e) {
            this.id = e.getAggregateId();
            this.customerId = e.getCustomerId();
            this.items = new ArrayList<>(e.getItems());
            this.status = "PLACED";
        } else if (event instanceof ItemAddedEvent e) {
            this.items.add(e.getItem());
        } else if (event instanceof OrderCancelledEvent e) {
            this.status = "CANCELLED";
        }
    }

    public List<DomainEvent> getUncommittedEvents() {
        return Collections.unmodifiableList(uncommittedEvents);
    }
    // 其他 getters...
}

// 使用 Project Reactor 的响应式流操作
public class ReactiveOrderService {
    private final OrderRepository orderRepository;
    private final NotificationService notificationService;

    public ReactiveOrderService(OrderRepository orderRepository,
                                 NotificationService notificationService) {
        this.orderRepository = orderRepository;
        this.notificationService = notificationService;
    }

    // 响应式查询：返回 Mono（0 或 1 个元素）
    public Mono<OrderDto> findOrder(String orderId) {
        return orderRepository.findById(orderId)
            .map(this::toDto)              // 转换为 DTO
            .switchIfEmpty(                // 空结果处理
                Mono.error(new OrderNotFoundException(orderId)));
    }

    // 响应式流查询：返回 Flux（0 到 N 个元素）
    public Flux<OrderDto> findOrdersByCustomer(String customerId) {
        return orderRepository.findByCustomerId(customerId)
            .filter(order -> "PLACED".equals(order.getStatus()))  // 过滤
            .map(this::toDto)                                     // 转换
            .take(100);                  // 限制数量（回压控制）
    }

    // 响应式命令：创建订单并发送通知
    public Mono<OrderResult> placeOrder(PlaceOrderCommand command) {
        return Mono.just(command)
            .flatMap(this::validateCommand)      // 校验
            .map(this::createOrder)              // 创建领域对象
            .flatMap(orderRepository::save)      // 持久化
            .flatMap(order ->                    // 发送通知
                notificationService.sendConfirmation(order)
                    .thenReturn(order))
            .map(this::toResult);                // 转换结果
    }
}

// 回压示例：消费者控制数据流速
public class BackpressureExample {
    public void processOrders(Flux<OrderEvent> eventStream) {
        eventStream
            .onBackpressureBuffer(1000)  // 缓冲区容量 1000
            .onBackpressureDrop()        // 超出缓冲区则丢弃
            .subscribe(
                event -> processEvent(event),       // 处理事件
                error -> handleError(error),        // 错误处理
                () -> handleComplete()              // 完成回调
            );
    }

    // 限速消费：每次只请求 10 个
    public void rateLimitedConsume(Flux<OrderEvent> eventStream) {
        eventStream
            .limitRate(10)              // 每次预取 10 个
            .subscribe(event -> processEvent(event));
    }
}

// 共享内存 + 锁的典型实现
public class ThreadSafeCounter {
    private int count = 0;
    private final Object lock = new Object();

    public void increment() {
        synchronized (lock) {
            count++;  // 临界区：同一时刻只有一个线程可进入
        }
    }

    public int getCount() {
        synchronized (lock) {
            return count;
        }
    }
}

// Akka Actor 示例
public class OrderActor extends AbstractActor {
    private final List<Order> orders = new ArrayList<>();
    // 状态仅在 Actor 内部，无需锁保护

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(PlaceOrderCommand.class, cmd -> {
                Order order = Order.create(cmd);
                orders.add(order);  // 安全：Actor 一次只处理一条消息
                getSender().tell(new OrderPlacedEvent(order.getId()), getSelf());
            })
            .match(CancelOrderCommand.class, cmd -> {
                orders.stream()
                    .filter(o -> o.getId().equals(cmd.getOrderId()))
                    .findFirst()
                    .ifPresent(order -> {
                        order.cancel(cmd.getReason());
                        getSender().tell(new OrderCancelledEvent(order.getId()), getSelf());
                    });
            })
            .build();
    }
}

// Go CSP 模型示例
func processOrders(orderCh <-chan Order, resultCh chan<- Result) {
    for order := range orderCh {  // 从通道接收订单
        result := process(order)  // 处理订单（无共享状态）
        resultCh <- result        // 将结果发送到通道
    }
}

func main() {
    orderCh := make(chan Order, 100)   // 缓冲通道
    resultCh := make(chan Result, 100)

    // 启动多个 Goroutine 并发处理
    for i := 0; i < 10; i++ {
        go processOrders(orderCh, resultCh)
    }

    // 发送订单
    for _, order := range fetchOrders() {
        orderCh <- order
    }
    close(orderCh)

    // 收集结果
    for result := range resultCh {
        fmt.Println(result)
    }
}

// 响应式流并发模型示例
public class ReactiveOrderProcessor {
    public Flux<OrderResult> processOrders(Flux<Order> orderStream) {
        return orderStream
            .flatMap(order ->                 // 并发处理每个订单
                Mono.fromCallable(() -> validate(order))
                    .subscribeOn(Schedulers.parallel()),  // 在并行调度器上执行
                10)                           // 最大并发度 = 10（回压控制）
            .flatMap(order ->                 // 持久化
                Mono.fromCallable(() -> save(order))
                    .subscribeOn(Schedulers.boundedElastic()),
                5)                            // 最大并发度 = 5
            .onErrorResume(err ->             // 错误恢复（韧性）
                Mono.just(OrderResult.failed(err.getMessage())));
    }
}