Flutter集群管理插件actor_cluster的使用

概述

⚠️ ALPHA ⚠️

此插件目前处于开发初期阶段。经常会有破坏性的变更。该插件尚未添加测试。

Actor Cluster

将多个actor系统连接到一个集群。

特性

• actor系统可以在不同的隔离区（isolate）和/或进程中运行。
• 使用自定义的序列化器/反序列化器在网络上传输消息。
• 当种子节点断开连接时，自动重新连接。
• 集群根据运行时指标决定在哪个节点上创建actor。
• 或者用户可以使用全限定的actor路径或节点标签来决定。

使用方法

示例代码

import 'dart:convert';
import 'dart:typed_data';

import 'package:actor_cluster/actor_cluster.dart';
import 'package:logging/logging.dart';

// 自定义的序列化器/反序列化器
class StringSerDes implements SerDes {
  [@override](/user/override)
  Object? deserialize(Uint8List data) {
    return utf8.decode(data); // 将字节数组解码为字符串
  }

  [@override](/user/override)
  Uint8List serialize(Object? message) {
    if (message is String) {
      return Uint8List.fromList(utf8.encode(message)); // 将字符串编码为字节数组
    }
    throw ArgumentError.value(message, 'message', 'not of type String');
  }
}

// 初始化日志记录
void initLogging() {
  Logger.root.level = Level.ALL;
  Logger.root.onRecord.listen((record) {
    if (record.loggerName.startsWith('actor://')) {
      print('${record.loggerName} ${record.level} ${record.message}');
    }
  });
}

void main(List<String> args) async {
  // 创建ActorCluster实例
  final clusterNode = ActorCluster(
    await readClusterConfigFromYaml('example/cluster.yaml'), // 从YAML文件读取集群配置
    await readNodeConfigFromYaml('example/${args[0]}.yaml'), // 从YAML文件读取节点配置
    StringSerDes(), // 使用自定义的序列化器
    initWorkerIsolate: (nodeId, workerId) => initLogging(), // 初始化日志记录
  );

  // 初始化集群
  await clusterNode.init(
    addActorFactories: (addActorFactory) {
      // 添加actor工厂
      addActorFactory(patternMatcher('/actor/1'), (path) {
        return (ActorContext context, Object? msg) async {
          final log = Logger(context.current.path.toString());
          final replyTo = context.replyTo;
          log.info('message: $msg'); // 打印消息
          log.info('correlationId: ${context.correlationId}'); // 打印相关ID
          log.info('replyTo: ${replyTo?.path}'); // 打印回复目标
          replyTo?.send(msg); // 发送消息
        };
      });

      addActorFactory(patternMatcher('/actor/2'), (path) {
        return (ActorContext context, Object? msg) async {
          final log = Logger(context.current.path.toString());
          log.info('message: $msg'); // 打印消息
          log.info('correlationId: ${context.correlationId}'); // 打印相关ID
          log.info('sender: ${context.sender?.path}'); // 打印发件人
          final actorRef = await context.lookupActor(actorPath('/actor/3')); // 查找actor
          actorRef?.send(msg); // 发送消息
        };
      });

      addActorFactory(patternMatcher('/actor/3'), (path) {
        return (ActorContext context, Object? msg) {
          final log = Logger(context.current.path.toString());
          log.info('message: $msg'); // 打印消息
          log.info('correlationId: ${context.correlationId}'); // 打印相关ID
          log.info('sender: ${context.sender?.path}'); // 打印发件人
        };
      });

      addActorFactory(patternMatcher('/foo'), (path) {
        return (ActorContext context, Object? msg) {
          final log = Logger(context.current.path.toString());
          log.info('message: $msg'); // 打印消息
        };
      });

      addActorFactory(patternMatcher('/bar'), (path) {
        return (ActorContext context, Object? msg) {
          final log = Logger(context.current.path.toString());
          log.info('message: $msg'); // 打印消息
        };
      });
    },
    initNode: (CreateActor createActor, List<String> tags) async {
      final actor = await createActor(actorPath('/foo'), 1000); // 创建actor
      await actor.send('test message'); // 发送消息
    },
    initCluster: (context) async {
      try {
        await context.createActor(actorPath('/bar', tag: 'foobar')); // 创建actor
        await context.createActor(actorPath('/bar', tag: 'foobar'));
        await context.createActor(actorPath('/bar', tag: 'foobar'));
      } catch (e) {
        print(e); // 打印异常信息
      }

      final actorRef1 = await context.createActor(Uri.parse('//node1/actor/1')); // 创建actor引用
      final actorRef2 = await context.createActor(Uri.parse('//node2/actor/2')); // 创建actor引用
      await context.createActor(Uri.parse('//node1/actor/3')); // 创建actor
      await actorRef1.send('hello cluster actor!', correlationId: '101', replyTo: actorRef2); // 发送消息

      print(await context.lookupActors(actorPath('/'))); // 查找所有actors
    },
  );
}