实战篇：Akka Actor 房间管理

概述

本篇把第 9 周的并发内容落地：用 Akka Actor 重构房间系统——玩家 Actor 管玩家、房间 Actor 管对局、匹配 Actor 管匹配，再用 Akka Cluster Sharding 把房间分片到多节点。通过完整代码展示 Actor 化房间的工程结构、消息通信与分片扩展。

---

一、案例设计与架构

1.1 Actor 划分

Actor 划分：
  PlayerActor     每玩家一个（状态：余额/背包/当前房间）
  RoomActor       每房间一个（状态：成员/对局/回合）
  MatchActor      按模式少量（状态：匹配队列）
  GatewayActor    网关桥接（接收网络消息路由给玩家）

通信链：
  玩家请求 → GatewayActor → PlayerActor → RoomActor
  房间事件 → RoomActor → 成员 PlayerActor

1.2 工程结构

com.game.akka
├── actor
│   ├── PlayerActor.java       // 玩家状态与操作
│   ├── RoomActor.java         // 房间与对局
│   ├── MatchActor.java        // 匹配池
│   └── GatewayActor.java      // 网关桥接
├── msg
│   ├── PlayerMsg.java         // 玩家消息（不可变）
│   ├── RoomMsg.java           // 房间消息
│   └── MatchMsg.java          // 匹配消息
├── model
│   ├── PlayerState.java       // 玩家状态（投影）
│   └── RoomState.java         // 房间状态（快照）
└── bootstrap
    └── GameClusterApp.java    // 启动 + 分片注册

---

二、PlayerActor 实现

public class PlayerActor extends AbstractActor {

    private final long playerId;
    private PlayerState state;      // 内存状态（单线程访问）

    public PlayerActor(long playerId) {
        this.playerId = playerId;
        this.state = snapshotStore.load(playerId);  // 启动恢复
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(JoinRoomCmd.class, this::onJoinRoom)
            .match(LeaveRoomCmd.class, this::onLeaveRoom)
            .match(PlayCardCmd.class, this::onPlayCard)
            .match(RoomEvent.class, this::onRoomEvent)
            .match(QuerySelfCmd.class, this::onQuerySelf)
            .build();
    }

    private void onJoinRoom(JoinRoomCmd cmd) {
        // 转发给房间 Actor（位置透明）
        ActorRef room = sharding.entityRefFor("RoomShard", cmd.roomId);
        room.tell(new PlayerJoinMsg(playerId), self());
    }

    private void onPlayCard(PlayCardCmd cmd) {
        ActorRef room = sharding.entityRefFor("RoomShard", state.getRoomId());
        room.tell(new PlayerActionMsg(playerId, cmd.card), self());
    }
}

---

三、RoomActor 实现

public class RoomActor extends AbstractActor {

    private final long roomId;
    private final RoomState state = new RoomState(roomId);

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(PlayerJoinMsg.class, this::onJoin)
            .match(PlayerActionMsg.class, this::onAction)
            .match(StartGameMsg.class, this::onStart)
            .match(ReconnectMsg.class, this::onReconnect)
            .build();
    }

    // 房间内所有操作单线程串行 → 无需加锁
    private void onJoin(PlayerJoinMsg msg) {
        if (state.addPlayer(msg.playerId)) {
            tellAll(new PlayerJoinedEvent(msg.playerId));
            persistState();          // 快照落盘（供故障恢复）
        }
    }

    private void onAction(PlayerActionMsg msg) {
        // 回合制校验 → 推进对局 → 广播结果
        TurnResult result = state.applyAction(msg.playerId, msg.card);
        tellAll(result);
        if (state.isFinished()) {
            settleAndStop(result);   // 结算后停止 Actor
        }
    }

    private void onReconnect(ReconnectMsg msg) {
        // 断线重连：返回房间快照
        sender().tell(state.snapshot(), self());
    }
}

---

四、MatchActor 与匹配流程

public class MatchActor extends AbstractActor {

    private final Queue<Long> waitQueue = new ArrayDeque<>();

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(ApplyMatchMsg.class, this::onApply)
            .match(CancelMatchMsg.class, this::onCancel)
            .build();
    }

    private void onApply(ApplyMatchMsg msg) {
        waitQueue.add(msg.playerId);
        if (waitQueue.size() >= 2) {
            long a = waitQueue.poll();
            long b = waitQueue.poll();
            long roomId = roomIdGen.nextId();
            // 创建房间 Actor（分片）并通知双方
            ActorRef room = sharding.entityRefFor("RoomShard", roomId);
            room.tell(new InitRoomMsg(a, b), self());
            playerActor(a).tell(new MatchFoundMsg(roomId), self());
            playerActor(b).tell(new MatchFoundMsg(roomId), self());
        }
    }
}

---

五、Cluster 分片扩展

public class GameClusterApp {

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("game-cluster", config());

        // 注册房间分片（按 roomId 路由）
        ClusterSharding.get(system).start(
            "RoomShard",
            RoomActor::props,
            ClusterShardingSettings.create(system),
            (msg) -> ((RoomMsg) msg).getRoomId()
        );

        // 匹配 Actor 每个节点一个（角色限定）
        system.actorOf(MatchActor.props(), "match-" + nodeId);
    }

    // 分片引用获取（自动路由到房间所在节点）
    public static ActorRef room(long roomId) {
        return ClusterSharding.get(system)
            .entityRefFor("RoomShard", roomId);
    }
}

扩展效果：
  加节点 → shard 自动 rebalance
  房间分散多节点 → 容量水平扩展
  节点宕机 → 房间在其他节点重建（快照恢复）

---

六、关键要点回顾

工程要点：
  状态内聚划分 Actor（玩家/房间/匹配）
  消息不可变，tell 异步通信
  房间单线程 → 无锁对局逻辑
  分片按 roomId 路由 → 水平扩展
  快照落盘 → 故障恢复

验证清单：
  同房间操作串行（无并发冲突）
  玩家跨节点请求正确路由
  节点下线房间可恢复
  匹配成功创建房间并通知
  断线重连返回快照
  空闲房间被回收（内存释放）

与 Netty 网关的衔接：
  网关收到消息 → 按 playerId 路由 PlayerActor
  PlayerActor 与 RoomActor 通过分片协作
  回执 → 网关 → 写回连接