Actor 模型

概念定义

共享状态要求线程用锁协调；Actor 模型用所有权作答：状态属于 actor、可达性经由消息、每个邮箱串行处理。争用消失，协调就消失。

工作机制

每个 actor 一次消费一条邮箱消息、无锁地变更私有状态。投递异步；顺序保证因运行时而异——Erlang 保证“同发送对”有序，其他未必。监督树让父级按策略重启失败的子 actor，把故障变成结构性事务。Orleans 的虚拟 actor 按 identity 按需激活，连放置都藏掉。

使用场景

大量独立有状态实体——会话、设备、游戏实体、工作流实例——按实体串行恰是期望语义、且按实体故障隔离要紧的地方。

常见误区

Actor 不是轻量线程而是“身份加邮箱”，执行骑在运行时调度器上。它消灭的是数据竞争不是逻辑竞争——actor 之间的消息顺序仍是设计工作。单例 actor 会让所有请求串行穿针；邮箱内部同步阻塞会锁死自己的邮箱并反向锁死被调者；一个业务操作拆成几十条消息的“话痨设计”是性能噩梦。突发下的邮箱溢出是其压力形态——要有界邮箱加背压信号；热 actor 须按键分区，拆分有状态 actor 还需要迁移设计。

代码示例

import { createClient } from 'redis';

const redis = createClient({ url: 'redis://127.0.0.1:6379' });
await redis.connect();

function randomToken() {
  return crypto.randomUUID();
}

async function acquireLock(key: string, ttlMs: number) {
  const token = randomToken();
  const ok = await redis.set('lock:' + key, token, { NX: true, PX: ttlMs });
  return ok ? token : null;
}

async function releaseLock(key: string, token: string) {
  const script = `
    if redis.call("GET", KEYS[1]) == ARGV[1] then
      return redis.call("DEL", KEYS[1])
    else
      return 0
    end
  `;
  return (await redis.eval(script, {
    keys: ['lock:' + key],
    arguments: [token],
  })) === 1;
}

async function withLock<T>(
  key: string, ttlMs: number, fn: () => Promise<T>,
): Promise<T | null> {
  const token = await acquireLock(key, ttlMs);
  if (token === null) return null;

  const renew = setInterval(() => {
    redis.expire('lock:' + key, ttlMs / 1000).catch(() => {});
  }, ttlMs / 3);

  try {
    return await fn();
  } finally {
    clearInterval(renew);
    await releaseLock(key, token);
  }
}

const result = await withLock('billing:nightly', 30_000, async () => {
  console.log('nightly billing started under lock');
  return { invoices: 42 };
});
console.log('lock result:', result?.invoices ?? 'busy');

use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel::<u32>();

    let producers: Vec<_> = (0..3)
        .map(|id| {
            let sender = sender.clone();
            thread::spawn(move || {
                for item in 1..=3 {
                    sender.send(id * 10 + item).expect("send failed");
                }
            })
        })
        .collect();
    drop(sender);

    let consumer = thread::spawn(move || {
        let mut total = 0u32;
        while let Ok(value) = receiver.recv() {
            total += value;
        }
        total
    });

    for handle in producers {
        handle.join().expect("producer panicked");
    }
    let total = consumer.join().expect("consumer panicked");
    println!("total = {total}");
}