登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›行为型模式-协作与交互机制

行为型模式-协作与交互机制

固拆棚 3 天前
行为型模式聚焦于对象间的行为交互，通过规范对象协作方式提升系统的灵活性与可扩展性。在分布式系统中，由于多节点异步通信、网络不可靠性及状态一致性挑战，行为型模式需针对分布式特性进行适应性设计。本文从观察者、策略、命令、责任链、状态五大核心行为型模式出发，系统解析其在分布式场景下的演化与实践。
一、观察者模式：分布式事件的发布 - 订阅机制

1.1 模式核心与分布式适配

观察者模式通过发布 - 订阅机制实现对象间的松耦合通信，在分布式系统中演变为跨节点的事件驱动架构（EDA），解决节点间异步通知问题。
1. 基于消息队列的分布式观察者
  1. // 事件定义（跨节点传输）
  2. public class OrderEvent implements Serializable {
  3.    private String eventId;
  4.    private Long orderId;
  5.    private OrderStatus status;
  6.    private LocalDateTime timestamp;
  7.    // getters/setters
  8. }
  10. // 抽象主题（事件发布者）
  11. public interface EventPublisher {
  12.    void publish(OrderEvent event);
  13. }
  15. // 具体发布者（订单服务）
  16. public class OrderService implements EventPublisher {
  17.    @Autowired
  18.    private KafkaTemplate<String, OrderEvent> kafkaTemplate;
  20.    @Override
  21.    public void publish(OrderEvent event) {
  22.        // 发布事件到Kafka主题
  23.        kafkaTemplate.send("order-events", event.getOrderId().toString(), event);
  24.    }
  25.    // 订单状态变更时发布事件
  26.    public void updateStatus(Long orderId, OrderStatus status) {
  27.        OrderEvent event = new OrderEvent(UUID.randomUUID().toString(), orderId, status, LocalDateTime.now());
  28.        publish(event);
  29.    }
  30. }
  31. // 抽象观察者（事件订阅者）
  32. public interface EventSubscriber {
  33.    void onEvent(OrderEvent event);
  34. }
  36. // 具体观察者（库存服务）
  37. public class InventorySubscriber implements EventSubscriber {
  38.    @KafkaListener(topics = "order-events")
  39.    @Override
  40.    public void onEvent(OrderEvent event) {
  41.        if (event.getStatus() == OrderStatus.PAID) {
  42.            // 订单支付后扣减库存
  43.            inventoryService.deduct(event.getOrderId());
  44.        }
  45.    }
  46. }
  48. // 具体观察者（积分服务）
  49. public class PointSubscriber implements EventSubscriber {
  50.    @KafkaListener(topics = "order-events")
  51.    @Override
  52.    public void onEvent(OrderEvent event) {
  53.        if (event.getStatus() == OrderStatus.PAID) {
  54.            // 订单支付后增加积分
  55.            pointService.add(event.getOrderId());
  56.        }
  57.    }
  58. }
复制代码
2. 分布式场景关键特性


  • 异步解耦：发布者无需知道订阅者存在（如订单服务不依赖库存 / 积分服务），降低节点耦合。
  • 可靠性保障：通过消息队列的持久化（如 Kafka 的日志存储）确保事件不丢失，应对节点宕机。
  • 广播能力：同一事件可被多个订阅者消费（如订单支付事件同时通知库存、积分、物流服务）。
1.2 与传统观察者的核心区别

维度单体观察者模式分布式观察者模式（事件驱动）通信方式内存直接调用消息队列异步通信（跨节点）可靠性依赖本地调用，失败直接抛出异常消息重试机制，支持死信队列处理失败事件订阅管理代码中直接注册观察者通过消息队列主题动态订阅（无需代码变更）二、策略模式：分布式场景的动态算法切换

2.1 模式核心与负载均衡策略

策略模式通过封装不同算法实现动态切换，在分布式系统中广泛应用于负载均衡、路由策略、序列化方式选择等场景。
1. 分布式负载均衡的策略设计
  1. // 策略接口（负载均衡算法）
  2. public interface LoadBalanceStrategy {
  3.    String selectInstance(List<String> instances);
  4. }
  6. // 具体策略1：轮询
  7. public class RoundRobinStrategy implements LoadBalanceStrategy {
  9.    private AtomicInteger index = new AtomicInteger(0);
  10.    
  11.    @Override
  12.    public String selectInstance(List<String> instances) {
  14.        if (instances.isEmpty()) return null;
  15.        int i = index.getAndIncrement() % instances.size();
  16.        return instances.get(i);
  17.    }
  18. }
  19. // 具体策略2：权重
  20. public class WeightedStrategy implements LoadBalanceStrategy {
  22.    @Override
  23.    public String selectInstance(List<String> instances) {
  24.        // 假设实例格式为"ip:port:weight"，解析权重并选择
  25.        int totalWeight = instances.stream()
  26.            .mapToInt(inst -> Integer.parseInt(inst.split(":")[2]))
  27.            .sum();
  28.        int random = new Random().nextInt(totalWeight);
  29.        int current = 0;
  30.        for (String inst : instances) {
  31.            int weight = Integer.parseInt(inst.split(":")[2]);
  32.            current += weight;
  33.            if (current > random) {
  34.                return inst;
  35.            }
  36.        }
  37.        return instances.get(0);
  38.    }
  39. }
  41. // 策略上下文（负载均衡器）
  42. public class LoadBalancer {
  43.    private LoadBalanceStrategy strategy;
  44.    // 动态设置策略（如从配置中心获取）
  45.    public void setStrategy(LoadBalanceStrategy strategy) {
  46.        this.strategy = strategy;
  47.    }
  48.    public String chooseInstance(String serviceName) {
  49.        // 从注册中心获取服务实例列表
  50.        List<String> instances = serviceDiscovery.getInstances(serviceName);
  51.        return strategy.selectInstance(instances);
  52.    }
  53. }
  55. // 使用示例
  56. public class ServiceConsumer {
  58.    public void invokeService() {
  59.        LoadBalancer balancer = new LoadBalancer();
  60.        // 从配置动态选择策略（如"weighted"）
  61.        String strategyType = Config.get("loadbalance.strategy");
  63.        balancer.setStrategy(StrategyFactory.create(strategyType));
  65.        String instance = balancer.chooseInstance("order-service");
  66.        // 调用选中的实例
  67.    }
  68. }
复制代码
2. 分布式场景价值


  • 动态适配：根据集群状态切换策略（如低负载时用轮询，高负载时用权重策略）。
  • 灰度发布：通过策略路由部分流量到新版本实例（如GrayStrategy只将 10% 流量路由到新实例）。
三、命令模式：分布式任务的异步执行与重试

3.1 模式核心与分布式命令

命令模式通过封装请求为对象，实现请求的异步执行、日志记录与重试，在分布式系统中演变为跨节点任务调度机制。
1. 分布式任务的命令设计
  1. // 命令接口
  3. public interface DistributedCommand extends Serializable {
  5.    String getCommandId(); // 唯一命令ID（用于幂等性）
  6.    CommandResult execute(); // 执行命令
  7.    CommandResult compensate(); // 补偿命令（失败时执行）
  8. }
  9. // 具体命令：订单支付命令
  10. public class PaymentCommand implements DistributedCommand {
  12.    private String commandId;
  13.    private Long orderId;
  14.    private BigDecimal amount;
  15.    @Override
  16.    public CommandResult execute() {
  17.        try {
  18.            // 调用支付服务
  19.            paymentService.pay(orderId, amount);
  20.            return CommandResult.success();
  21.        } catch (Exception e) {
  22.            return CommandResult.failure(e.getMessage());
  23.        }
  24.    }
  26.    @Override
  27.    public CommandResult compensate() {
  29.        // 支付失败时执行退款
  30.        return paymentService.refund(orderId) ? CommandResult.success() : CommandResult.failure("退款失败");
  31.    }
  32.    // getters/setters
  33. }
  35. // 命令调用者（任务调度器）
  36. public class CommandScheduler {
  38.    @Autowired
  39.    private KafkaTemplate<String, DistributedCommand> kafkaTemplate;
  41.    @Autowired
  42.    private CommandRepository repository; // 命令日志存储
  44.    // 发送命令到执行节点
  45.    public void schedule(DistributedCommand command) {
  46.        // 保存命令日志（用于重试和恢复）
  47.        repository.save(new CommandLog(command.getCommandId(), command, CommandStatus.PENDING));
  48.        // 发送到命令主题
  49.        kafkaTemplate.send("command-topic", command.getCommandId(), command);
  50.    }
  51.    // 处理命令结果
  52.    public void handleResult(String commandId, CommandResult result) {
  54.        CommandLog log = repository.findById(commandId);
  56.        if (result.isSuccess()) {
  57.            log.setStatus(CommandStatus.SUCCESS);
  58.        } else {
  59.            log.setStatus(CommandStatus.FAILED);
  60.            // 失败重试（最多3次）
  61.            if (log.getRetryCount() < 3) {
  62.                log.incrementRetryCount();
  63.                kafkaTemplate.send("command-topic", commandId, log.getCommand());
  64.            }
  65.        }
  66.        repository.update(log);
  67.    }
  68. }
  70. // 命令执行者（工作节点）
  71. public class CommandWorker {
  73.    @KafkaListener(topics = "command-topic")
  74.    public void executeCommand(ConsumerRecord<String, DistributedCommand> record) {
  76.        DistributedCommand command = record.value();
  77.        CommandResult result = command.execute();
  79.        // 发送执行结果
  80.        commandScheduler.handleResult(command.getCommandId(), result);
  81.    }
  82. }
复制代码
2. 分布式场景优势


  • 异步执行：命令通过消息队列异步发送到工作节点，调用者无需等待执行完成。
  • 故障恢复：命令日志记录执行状态，节点宕机后可从日志恢复未完成命令。
  • 幂等设计：通过commandId确保重复执行（如消息重试）不会产生副作用。
四、责任链模式：分布式请求的链式处理

4.1 模式核心与 API 网关过滤链

责任链模式通过多个处理器依次处理请求，在分布式系统中广泛应用于 API 网关的请求过滤、分布式事务的阶段处理等场景。
1. API 网关的责任链设计
  1. // 处理器接口
  2. public interface GatewayFilter {
  4.    void doFilter(GatewayRequest request, GatewayResponse response, GatewayFilterChain chain);
  5. }
  7. // 具体过滤器1：认证过滤
  8. public class AuthFilter implements GatewayFilter {
  10.    @Override
  11.    public void doFilter(GatewayRequest request, GatewayResponse response, GatewayFilterChain chain) {
  13.        String token = request.getHeader("Authorization");
  15.        if (token == null || !authService.validate(token)) {
  16.            response.setStatusCode(401);
  17.            response.setBody("Unauthorized");
  18.            return; // 终止链
  19.        }
  20.        chain.doFilter(request, response); // 继续下一个过滤器
  21.    }
  22. }
  24. // 具体过滤器2：限流过滤
  25. public class RateLimitFilter implements GatewayFilter {
  27.    private RedisTemplate<String, Integer> redisTemplate;
  28.    
  29.    @Override
  30.    public void doFilter(GatewayRequest request, GatewayResponse response, GatewayFilterChain chain) {
  32.        String clientIp = request.getClientIp();
  33.        String key = "rate_limit:" + clientIp;
  34.        // 使用Redis实现限流
  35.        Long count = redisTemplate.opsForValue().increment(key, 1);
  37.        if (count == 1) {
  38.            redisTemplate.expire(key, 1, TimeUnit.MINUTES);
  39.        }
  40.        if (count > 100) { // 每分钟最多100次请求
  41.            response.setStatusCode(429);
  42.            response.setBody("Too Many Requests");
  43.            return;
  44.        }
  45.        chain.doFilter(request, response);
  46.    }
  47. }
  49. // 过滤器链
  50. public class DefaultGatewayFilterChain implements GatewayFilterChain {
  52.    private List<GatewayFilter> filters;
  53.    private int index = 0;
  55.    public DefaultGatewayFilterChain(List<GatewayFilter> filters) {
  56.        this.filters = filters;
  57.    }
  59.    @Override
  60.    public void doFilter(GatewayRequest request, GatewayResponse response) {
  62.        if (index < filters.size()) {
  63.            GatewayFilter filter = filters.get(index++);
  64.            filter.doFilter(request, response, this);
  65.        }
  66.    }
  67. }
  69. // API网关
  70. public class ApiGateway {
  72.    public GatewayResponse handleRequest(GatewayRequest request) {
  74.        GatewayResponse response = new GatewayResponse();
  75.        // 构建过滤器链（认证→限流→路由）
  76.        List<GatewayFilter> filters = Arrays.asList(
  77.            new AuthFilter(),
  78.            new RateLimitFilter(),
  79.            new RouteFilter()
  80.        );
  82.        new DefaultGatewayFilterChain(filters).doFilter(request, response);
  84.        return response;
  85.    }
  86. }
复制代码
2. 分布式场景价值


  • 横切逻辑复用：认证、限流等逻辑集中在网关，无需每个服务重复实现。
  • 动态扩展：通过配置中心动态增删过滤器（如临时添加维护模式过滤器）。
五、状态模式：分布式系统的状态机管理

5.1 模式核心与订单状态流转

状态模式通过封装对象状态及其转换逻辑，在分布式系统中用于管理跨节点的状态一致性（如订单状态、工作流状态）。
1. 分布式订单状态机
  1. // 状态接口
  2. public interface OrderState {
  4.    void pay(OrderContext context); // 支付操作
  5.    void ship(OrderContext context); // 发货操作
  6.    void complete(OrderContext context); // 完成操作
  7.    OrderStatus getStatus(); // 获取当前状态
  9. }
  11. // 具体状态：待支付
  12. public class PendingState implements OrderState {
  14.    @Override
  15.    public void pay(OrderContext context) {
  17.        // 状态转换：待支付→已支付
  18.        context.setState(new PaidState());
  20.        // 发布状态变更事件（通知其他节点）
  21.        context.publishEvent(OrderStatus.PAID);
  23.    }
  25.    @Override
  26.    public void ship(OrderContext context) {
  28.        throw new IllegalStateException("未支付订单不能发货");
  29.    }
  31.    @Override
  32.    public void complete(OrderContext context) {
  34.        throw new IllegalStateException("未支付订单不能完成");
  35.    }
  37.    @Override
  38.    public OrderStatus getStatus() { return OrderStatus.PENDING; }
  40. }
  41. // 具体状态：已支付（其他状态略）
  42. public class PaidState implements OrderState { /* 实现发货等操作 */ }
  43. // 上下文：订单状态管理器
  45. public class OrderContext {
  47.    private OrderState currentState;
  48.    private Long orderId;
  49.    private EventPublisher eventPublisher;
  50.    public OrderContext(Long orderId) {
  51.        this.orderId = orderId;
  53.        this.currentState = new PendingState(); // 初始状态
  55.    }
  57.    // 委托状态操作
  58.    public void pay() { currentState.pay(this); }
  60.    public void ship() { currentState.ship(this); }
  62.    public void complete() { currentState.complete(this); }
  64.    // 发布状态事件（跨节点同步）
  65.    public void publishEvent(OrderStatus status) {
  67.        eventPublisher.publish(new OrderEvent(orderId, status));
  69.    }
  71.    // 设置状态（仅允许状态内部调用）
  72.    void setState(OrderState state) { this.currentState = state; }
  74. }
  76. // 使用示例
  77. public class OrderController {
  79.    @PostMapping("/orders/{id}/pay")
  80.    public void payOrder(@PathVariable Long id) {
  81.        OrderContext context = orderContextManager.get(id);
  82.        context.pay(); // 状态转换由状态机管理
  83.    }
  84. }
复制代码
2. 分布式场景挑战与解决


  • 状态一致性：通过事件发布同步状态变更（如订单服务状态变更后，通知库存服务）。
  • 并发安全：状态转换加分布式锁（如 Redis 锁），防止并发操作导致状态错乱。
六、面试高频问题深度解析

6.1 基础概念类问题

Q：分布式环境下的观察者模式与传统观察者模式有何本质区别？如何保证事件不丢失？
A：

  • 本质区别：
    传统观察者是进程内同步调用（如内存中的发布 - 订阅），分布式观察者基于消息队列异步通信，跨节点、跨进程。
  • 不丢失保证：

  • 消息持久化：Kafka 将消息写入磁盘，确保 broker 宕机后数据不丢失。
  • 确认机制：消费者处理完成后发送 ACK（如 Kafka 的 offset 提交），未确认则重试。
  • 死信队列：多次重试失败的事件进入死信队列，人工干预处理。
Q：策略模式在分布式负载均衡中的应用？如何动态切换负载均衡策略？
A：

  • 应用场景：
    策略模式封装轮询、权重、IP 哈希等负载均衡算法，LoadBalancer通过setStrategy()动态选择算法。
  • 动态切换：

  • 配置中心（如 Nacos）存储策略类型（如weighted）。
  • 客户端监听配置变更，调用setStrategy()更新策略。
  • 示例：流量高峰时切换到权重策略（优先调度到高配节点），低谷时用轮询策略。
6.2 实战设计类问题

Q：如何用责任链模式设计一个支持多租户的 API 网关权限系统？
A：

  • 过滤器链设计：


  • TenantFilter：解析请求中的租户 ID，验证租户合法性。
  • AuthFilter：验证租户内用户的令牌有效性。
  • PermissionFilter：检查用户是否有权限访问当前接口（结合租户 + 用户角色）。
  • RateLimitFilter：按租户限制 API 调用频率。

  • 执行逻辑：
    网关接收请求后，依次执行过滤器链，任何过滤器失败则返回对应错误（如 401/403）。
  • 租户隔离：
    每个过滤器通过request.getTenantId()获取租户上下文，确保权限校验在租户维度隔离。
Q：分布式状态机如何保证跨节点的状态一致性？
A：

  • 状态事件同步：状态变更时发布全局事件（如OrderStatusChangedEvent），所有节点消费事件更新本地状态。
  • 分布式锁：状态转换操作加锁（如 Redis 锁），防止并发修改导致状态冲突。
  • 状态校验：节点启动时从事件日志重建状态（如从 Kafka 消费历史事件恢复最新状态）。
总结：行为型模式的分布式设计原则

核心选型策略

分布式场景推荐模式核心解决问题跨节点异步通知观察者模式（消息队列）节点解耦，事件驱动协作动态算法切换（负载均衡等）策略模式算法与使用分离，支持动态适配分布式任务调度与重试命令模式任务异步执行，支持补偿与幂等性网关过滤、请求处理链责任链模式横切逻辑复用，动态扩展过滤器跨节点状态流转（订单等）状态模式状态转换逻辑封装，保证状态一致性分布式适配要点


  • 异步优先：尽量采用异步通信（消息队列）减少节点阻塞，应对网络延迟。
  • 幂等设计：所有跨节点操作需保证幂等性（如命令 ID、事件 ID），防止重试导致副作用。
  • 容错机制：结合重试、超时控制、熔断降级，应对网络分区与节点故障。
通过掌握行为型模式在分布式系统中的适配逻辑，不仅能在面试中清晰解析跨节点协作问题，更能在实际架构中设计松耦合、高可用的分布式系统，体现高级程序员对复杂系统的设计能力。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
固拆棚
3 天前

0

粉丝关注

18

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫