如何阻止/减缓CAF生产者Actor淹没消费者邮箱？

一、Actor邮箱流控实现 #

1. 正确获取邮箱消息数并实现暂停/恢复 #

之前用self->mailbox.size()出现周期异常，是因为该接口属于内部实现，调用时会触发线程同步逻辑，打乱生产者的调度周期。推荐用Actor间消息传递的反馈方式，同时使用安全的邮箱状态查询接口：

// 消费者定期向生产者汇报邮箱负载
behavior consumer(event_based_actor* self, actor producer) {
  // 每50ms发送一次当前邮箱消息数
  self->schedule_periodically(50ms, [self, producer] {
    self->send(producer, self->mailbox_count());
  });
  return {
    [self](image_msg& img) {
      // 模拟耗时图像处理
      std::this_thread::sleep_for(200ms);
    }
  };
}

// 生产者根据反馈控制发送节奏
behavior producer(event_based_actor* self, actor consumer, size_t max_mailbox) {
  bool is_paused = false;
  // 初始按100ms周期尝试发送
  self->schedule_periodically(100ms, [self, consumer, &is_paused] {
    if (!is_paused) {
      self->send(consumer, image_msg{});
    }
  });
  return {
    [self, max_mailbox, &is_paused](size_t current_count) {
      // 超过阈值暂停，低于阈值的一半恢复
      if (current_count >= max_mailbox) {
        is_paused = true;
      } else if (current_count < max_mailbox / 2) {
        is_paused = false;
      }
    }
  };
}

关于caf.actor.mailbox-size编译报错：该系统指标需通过actor_system::metrics()接口获取，且需要启用CAF的metrics模块，不如直接用mailbox_count()接口简洁可靠。

2. 简化配置读取 #

无需手动解析配置文件，直接用CAF原生接口读取自定义配置项：

actor_system_config cfg;
cfg.load("app_config.ini");
// 读取配置，默认值设为5
auto max_mailbox = cfg.get_or("stream.max-mailbox-size", 5);
// 启动Actor时传入配置值
auto consumer = sys.spawn(consumer, producer);
auto producer = sys.spawn(producer, consumer, max_mailbox);

二、caf::streams适配相机图像流场景 #

完全适用，CAF流原生支持背压机制，正好匹配“生产者固定速率推流、消费者耗时处理”的需求：

• 无需手动实现邮箱检测，流框架会自动根据消费者的处理能力调节发送速率，避免邮箱溢出。
• 消费者处理耗时任务时，会自动暂停拉取动作，生产者将停止发送直到消费者就绪。
• 简化实现代码：

// 图像生产者流源
behavior image_producer(event_based_actor* self) {
  return {
    [self](stream_init<image_msg>& si) {
      auto out = si.out();
      // 每100ms生成一张图像
      self->schedule_periodically(100ms, [self, out] {
        if (!out.closed()) {
          out.push(image_msg{});
        }
      });
      return out;
    }
  };
}

// 图像消费者流Sink
behavior image_consumer(event_based_actor* self) {
  return {
    [self](stream_sink_init<image_msg>& si) {
      auto in = si.in();
      in.set_behavior([](image_msg& img) {
        // 模拟耗时处理
        std::this_thread::sleep_for(200ms);
        return task_result::ok;
      });
      return in;
    }
  };
}

// 连接流
auto producer = sys.spawn(image_producer);
auto consumer = sys.spawn(image_consumer);
self->send(producer, stream_init<image_msg>{}.to(consumer));

内容的提问来源于stack exchange，提问作者luis castro