Akka Typed项目中如何正确查找、生成Actor并发送消息？

你的第二种写法思路可行，但存在几个关键问题，下面逐一分析并给出改进方案：

现有写法的问题 #

1. 未处理异常分支：WrappedFindResult包含Throwable failure字段，但代码直接访问result.listing()，如果AskPattern.ask超时或失败，listing会是null，调用getServiceInstances会触发空指针异常。
2. Worker未注册到Receptionist：找不到Worker时创建的匿名Actor没有注册到Receptionist，导致后续查找仍会重复创建新的Worker，造成资源浪费。
3. Behavior切换逻辑错误：处理完WrappedFindResult后返回this，也就是当前临时的receive Behavior，后续其他Command消息会被忽略，因为这个Behavior只处理WrappedFindResult类型。

改进后的代码示例 #

private Behavior<Command> tellToWorker(ServiceKey<Worker.Command> serviceKey, Worker.Command command, Behavior<Command> originalBehavior) {
    record WrappedFindResult(Receptionist.Listing listing, Throwable failure) implements Command {}

    var system = getContext().getSystem();
    getContext().pipeToSelf(
            AskPattern.ask(
                    system.receptionist(),
                    replyTo -> Receptionist.find(serviceKey, replyTo),
                    Duration.ofSeconds(1),
                    system.scheduler()),
            WrappedFindResult::new);

    return Behaviors.receive(Command.class)
            .onMessage(WrappedFindResult.class, result -> {
                // 处理异常情况
                if (result.failure() != null) {
                    // 这里可以添加日志、重试逻辑或降级处理
                    System.err.println("查找Worker失败: " + result.failure().getMessage());
                    return originalBehavior;
                }

                // 查找或创建Worker
                WorkerRef worker = result.listing().getServiceInstances(serviceKey)
                        .stream()
                        .findFirst()
                        .orElseGet(() -> {
                            WorkerRef newWorker = getContext().spawnAnonymous(Worker.create());
                            // 将新创建的Worker注册到Receptionist
                            system.receptionist().tell(Receptionist.register(serviceKey, newWorker));
                            return newWorker;
                        });

                // 发送命令
                worker.tell(command);

                // 回到原来的Behavior，确保后续消息能正常处理
                return originalBehavior;
            })
            // 临时Behavior收到其他消息时，转发给原Behavior处理
            .onAnyMessage(msg -> {
                originalBehavior = Behaviors.same(originalBehavior, msg);
                return originalBehavior;
            })
            .build();
}

进一步优化建议 #

• 使用Receptionist订阅替代单次查找：如果需要持续关注Worker实例的变化，可以通过Receptionist.subscribe监听服务注册/注销事件，避免每次发送消息都发起查找请求。示例如下：

  // 在Behavior初始化时订阅服务
  getContext().getSystem().receptionist().tell(
          Receptionist.subscribe(serviceKey, getContext().messageAdapter(Receptionist.Listing.class, ListingUpdate::new)));
  
  // 定义ListingUpdate作为Command的实现类
  record ListingUpdate(Receptionist.Listing listing) implements Command {}
  
  // 在Behavior中维护Worker实例缓存，处理ListingUpdate时更新缓存
  private WorkerRef cachedWorker;
  
  .onMessage(ListingUpdate.class, update -> {
      cachedWorker = update.listing().getServiceInstances(serviceKey)
              .stream()
              .findFirst()
              .orElseGet(() -> {
                  WorkerRef newWorker = getContext().spawnAnonymous(Worker.create());
                  getContext().getSystem().receptionist().tell(Receptionist.register(serviceKey, newWorker));
                  return newWorker;
              });
      return Behaviors.same();
  })
  

  后续发送消息时直接使用缓存的cachedWorker即可。

• 避免匿名Actor（可选）：如果需要追踪Worker实例，可以使用命名Actor（spawn而非spawnAnonymous），方便日志和监控。

• 添加超时和重试策略：对于查找失败的情况，可以使用Behaviors.withTimers实现定时重试逻辑。

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