C# 实现 Actor并发模型 (案例版)

啥是Actor模型

Actor (英语翻译演员) 这个概念要回溯到面向对象程序设计的本身上来，更偏向于现实世界，现实世界就是由单个个体（人）与其他个体或(人）通讯（消息）组成的现实世界，所以，它的好处是显而易见的，生活就是Actor。

现有的面向对象编程模型是基于内存共享线程模型的，而Actor是基于消息模型的，Actor 之间是完全隔离的，不会共享任何变量。

基于内存，那么，内存会溢出，异常，基于消息的话，则没有这种困扰。

又由于它变相的来讲是单线程的，自己处理自己的事务，又实现了与其他业务的隔离。

Erlang语言，天生支持Actor模型，是一个好语言啊。

Actor 的优势

1. Actor不共享状态
2. 高并发，无锁（无共享状态）
3. Actor效率高（同一时刻只处理一个任务，属于单线程处理）

Actor 框架有

1. Orleans
2. Dapr
3. Akka.NET
4. Proto.Actor

Actor模型的实现

为啥要实现Actor模型，一个是为了更深入的了解它，一个是想实现一下。

可能在某些地方直接就用了。没必要搞那么复杂引用。

Actor模型的原理

按照自己的理解画了一下。

简单来讲，就是各个服务都可以投递消息到Actor实例里，Actor会从邮箱里把消息取出来，然后，消费掉。这么简单的一件事情。

Actor 代码逻辑的实现

IActor.cs

/// <summary>
/// 无锁并行编程模型（暂时用来处理串行任务，任务串行执行）
/// </summary>
public interface IActor
{/// <summary>/// 增加消息/// </summary>/// <returns></returns>bool AddMsg(object message);/// <summary>/// 启动服务/// </summary>/// <returns></returns>Task Start();/// <summary>/// 停止服务运行，等待毫秒数/// </summary>/// <param name="WatingTimeout"></param>/// <returns></returns>bool Stop(int WatingTimeout);
}

Actor.cs

/// <summary>
/// Actor抽象
/// </summary>
public abstract class Actor : IDisposable, IActor
{public Actor(string name){Name = name;MailBox = new BlockingCollection<object>();}/// <summary>/// 名称/// </summary>public string Name { get; set; }/// <summary>/// 是否启用/// </summary>public bool Active { get; private set; }/// <summary>/// 是否长时间运行。长时间运行任务使用独立线程，默认true/// </summary>public bool LongRunning { get; set; } = true;/// <summary>/// 处理的消息邮箱/// </summary>public BlockingCollection<object> MailBox { get; set; }/// <summary>/// 内置任务/// </summary>private Task _task;public virtual Task Start(){if (Active) return _task;Active = true;// 启动异步if (_task == null){lock (this){if (_task == null){_task = Task.Factory.StartNew(DoActorWork, LongRunning ? TaskCreationOptions.LongRunning : TaskCreationOptions.None);}}}return _task;}public virtual bool Stop(int WatingTimeout = 100){MailBox?.CompleteAdding();Active = false;if (WatingTimeout == 0 || _task == null) return true;return _task.Wait(WatingTimeout);}public virtual bool AddMsg(object message){// 自动开始if (!Active){Start();}if (!Active){return false;}MailBox.Add(message);return true;}/// <summary>/// 循环消费消息/// </summary>private void DoActorWork(){while (!MailBox.IsCompleted){try{var ctx = MailBox.Take();var task = ProcessAsync(ctx);if (task != null){task.Wait();}}catch (InvalidOperationException) { }catch (Exception ex){Console.WriteLine($"DoActorWork Error : {ex.Message}");}}Active = false;}/// <summary>/// 处理消息/// </summary>/// <returns></returns>public abstract Task ProcessAsync(object msg);public void Dispose(){try{Stop(100);}catch (Exception){}while (MailBox?.TryTake(out _) == true) { }MailBox = null;}
}

测试代码

static void Main(string[] args)
{Console.Title = "Actor Demo by 蓝创精英团队";//实现一个加法逻辑//a累加到100，就发送消息到 b里，让b 输出。var write = new WriteActor();var User = new AccumulationActor(write);for (int i = 0; i < 20; i++){User.AddMsg(i * 30);}Thread.Sleep(2000);write.Stop();User.Stop();//释放资源Console.WriteLine("示例完毕!");Console.ReadLine();
}