3.actor模型的根本搭建（c#代码）

本文看了博客C# 实现 Actor并发模型 (案例版)_51CTO博客_actor并发模型，这里作为条记用，该博客内容写的很具体，这里根本上没有改动。
起首，本文的目次如下：

每个cs文件的代码都有具体的注释，具体代码如下：
1. IActor.cs代码：

1. using System;
2. using System.Collections.Generic;
3. using System.Linq;
4. using System.Text;
5. using System.Threading.Tasks;
7. namespace ConsoleApp3.core
8. {
9.     /// <summary>
10.     /// 无锁并行编程模型（暂时用来处理串行任务，任务串行执行）
11.     /// </summary>
12.     public interface IActor
13.     {
15.         bool AddMsg(object message); //增加消息
17.         Task Start(); // 启动服务
19.         bool Stop(int WaitingTimeout); // 停止服务运行，等待毫秒数
21.     }
22. }

复制代码

2. Actor.cs代码

1. using System;
2. using System.Collections.Concurrent;
3. using System.Collections.Generic;
4. using System.Linq;
5. using System.Text;
6. using System.Threading.Tasks;
8. namespace ConsoleApp3.core
9. {
10.     public abstract class Actor : IDisposable, IActor
11.     {
12.         public string Name { get; set; }//名称
14.         public bool Active { get; private set; } // 是否启用
16.         public bool LongRunning { get; set; } = true; // 是否长时间运行。长时间运行任务使用独立线程，默认true
18.         public BlockingCollection<object> MailBox { get; set; } // 处理的消息邮箱
20.         private Task _task; // 内置任务
22.         public Actor(string name)
23.         {
24.             Name = name;
25.             MailBox = new BlockingCollection<object>();
26.         }
28.         // 开始任务
29.         public virtual Task Start()
30.         {
31.             if (Active) return _task;
32.             Active = true;
33.             // 启动异步
34.             if (_task == null)
35.             {
36.                 lock(this)
37.                 {
38.                     if (_task == null)
39.                     {
40.                         // Task.Factory.StartNew: 线程的使用
41.                         // 如果LongRunning为真，则返回 TaskCreationOptions.LongRunning，否则返回TaskCreationOptions.None
42.                         _task = Task.Factory.StartNew(DoActorWork, LongRunning ? TaskCreationOptions.LongRunning : TaskCreationOptions.None);
43.                     }
44.                 }
45.             }
46.             return _task;
47.         }
49.         // 停止任务
50.         public virtual bool Stop(int WaitingTimeout = 100)
51.         {
53.             //BlockingCollection?.CompleteAdding()它的作用是通知集合不再接受新的元素添加。调用此方法后，尝试向集合中添加元素的操作将会抛出异常，并且当集合为空时，
54.             //尝试从集合中移除元素的操作不会等待，而是立即返回。
55.             MailBox?.CompleteAdding();
56.             Active = false;
57.             if (WaitingTimeout == 0 || _task == null) return true;
59.             return _task.Wait(WaitingTimeout);
60.         }
62.         // 给邮件添加消息
63.         public virtual bool AddMsg(object message)
64.         {
65.             // 自动开始
66.             if (!Active) { Start(); }
67.             if (!Active) { return false; }
68.             MailBox.Add(message);
69.             return true;
70.         }
72.         // 循环消息
73.         private void DoActorWork()
74.         {
75.             //BlockingCollection.IsCompleted：指集合是否已经完成添加并且是否已经消耗了所有数据。换句话说，它返回 true 的条件是集合已经调用了 CompleteAdding() 并且集合为空。
76.             //因此，IsCompleted 包含了集合为空的判断
77.             while(!MailBox.IsCompleted)
78.             {
79.                 try
80.                 {
81.                     //BlockingCollection.Take()：用于从集合中移除并返回一个元素。如果集合为空，调用线程会被阻塞，直到集合中有元素可用。
82.                     var ctx = MailBox.Take();
83.                     var task = ProcessAsync(ctx);
84.                     if (task != null)
85.                     {
86.                         task.Wait();
87.                     }
88.                 }
89.                 catch(InvalidOperationException) { }
90.                 catch (Exception ex)
91.                 {
92.                     Console.WriteLine($"DoActorWork Error : {ex.Message}");
93.                 }
94.             }
95.             Active = false;
96.         }
98.         // 处理消息
99.         // abstract：是定义了一个抽象类
100.         public abstract Task ProcessAsync(object msg);
102.         public void Dispose()
103.         {
104.             try {Stop(100); }
105.             catch (Exception) { }
106.             // BlockingCollection?.TryTake(out _)：尝试从MailBox中移除一个元素。如果集合为空，该方法会立即返回false；如果集合不为空，则尝试移除一个元素并返回true。
107.             // 如果操作成功，还可以通过out参数获取移除的元素。
108.             while (MailBox?.TryTake(out _) == true) { }
109.             MailBox = null;
110.         }
112.     }
113. }

复制代码

3.WriteActor.cs代码

1. using ConsoleApp3.core;
2. using System;
3. using System.Collections.Generic;
4. using System.Linq;
5. using System.Text;
6. using System.Threading.Tasks;
8. namespace ConsoleApp3
9. {
10.     public class WriteActor: Actor
11.     {
12.         public WriteActor() : base(nameof(WriteActor)) { }
14.          处理信息
15.         public override Task ProcessAsync(object msg)
16.         {
17.             try { Console.WriteLine($"输出 {this.Name} :{msg}");}
18.             catch(Exception e) { Console.WriteLine($"业务处理异常:{e.Message}"); }
19.             return Task.CompletedTask;
20.         }
21.     }
22. }

复制代码

4.AccumulationActor.cs代码

1. using ConsoleApp3.core;
2. using System;
3. using System.Collections.Generic;
4. using System.Linq;
5. using System.Text;
6. using System.Threading.Tasks;
8. namespace ConsoleApp3
9. {
10.     public class AccumulationActor : Actor
11.     {
12.         private int Count = 0;
13.         private IActor actor;
14.         public AccumulationActor(IActor actor) : base(nameof(AccumulationActor))
15.         {
16.             Count = 0;
17.             this.actor = actor;
18.             //Console.WriteLine(nameof(AccumulationActor)); // 输出基类名称
19.         }
20.         // 处理信息
21.         public override Task ProcessAsync(object msg)
22.         {
23.             try
24.             {
25.                 var msgNumber = (int)(msg);
26.                 Count += msgNumber;
27.                 Console.WriteLine($"处理{this.Name} :{msg} ，累积总数:{Count}");
29.                 if (Count >= 100)
30.                 {
31.                     this.actor.AddMsg(Count);
32.                     Count = 0;
33.                 }
34.             }
35.             catch(Exception e) { Console.WriteLine($"业务处理异常:{e.Message}"); }
37.             return Task.CompletedTask;
38.         }
39.     }
40. }

复制代码

5.program.cs代码

1. using System;
2. using System.Collections.Generic;
3. using System.Linq;
4. using System.Text;
5. using System.Threading;
6. using System.Threading.Tasks;
8. namespace ConsoleApp3
9. {
10.     class Program
11.     {
12.         static void Main(string[] args)
13.         {
15.             Console.Title = "Actor Demo of wang";
17.             //实现一个加法逻辑
18.             //a累加到100，就发送消息到 b里，让b 输出
20.             var write = new WriteActor();
21.             var User = new AccumulationActor(write);
22.             for (int i = 0; i < 20; i++) { User.AddMsg(i * 30); }
24.             Thread.Sleep(2000);
25.             write.Stop();
26.             User.Stop();
27.             //释放资源
28.             Console.WriteLine("示例完毕!");
29.             Console.ReadLine();
30.         }
31.     }
32. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。