.Net 5 Actor 并发模型

清疚 2022-08-28 12:54 200阅读 0赞

是一个利器，用来实现 模型通讯，并发执行。

/// <summary>
        /// 无锁并行编程模型（暂时用来处理串行任务，任务串行执行）
        /// </summary>
        public interface IActor
        {
            /// <summary>
            /// 增加消息
            /// </summary>
            /// <returns></returns>
            (bool success, int size) AddMsg(object message);
            /// <summary>
            /// 启动服务
            /// </summary>
            /// <returns></returns>
            Task Start();
            /// <summary>
            /// 停止服务运行，等待毫秒数
            /// </summary>
            /// <param name="WatingTimeout"></param>
            /// <returns></returns>
            bool Stop(int WatingTimeout);
        }
        public abstract class Actor : IDisposable, IActor
        {
            readonly ILog log = Logger.LogManager.GetLogger("DefaultLog");
            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)
            {
                MailBox?.CompleteAdding();
                Active = false;
                if (WatingTimeout == 0 || _task == null) return true;
    
                return _task.Wait(WatingTimeout);
            }
            public virtual (bool success, int size) AddMsg(object message)
            {
                // 自动开始
                if (!Active) Start();
    
                if (!Active)
                {
                    return (false, MailBox.Count);
                }
                MailBox.Add(message);
    
                return (true, MailBox.Count);
            }
            /// <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)
                    {
                        log.Error(ex.Message, ex);
                    }
                }
    
                Active = false;
            }
            /// <summary>
            /// 处理消息
            /// </summary>
            /// <returns></returns>
            public abstract Task ProcessAsync(object msg);
            public void Dispose()
            {
                while (MailBox?.TryTake(out _) == true) { }
                MailBox = null;
            }
        }

/// <summary>
        /// 数据库处理
        /// </summary>
        public class ComputerActor : Actor
        {
            /// <summary>
            /// 串行处理
            /// </summary>
            /// <param name="TableName">表名</param>
            public ComputerActor(string TableName) : base(TableName)
            {
            }
            /// 如果已经有数据，那就不增加数据处理了
            /// </summary>
            /// <returns></returns>
            public override (bool success, int size) AddMsg(object message)
            {
                if (MailBox.Count > 0)
                {
                    return (false, MailBox.Count);
                }
                return base.AddMsg(message);
            }
            /// <summary>
            /// <summary>
            /// 处理信息
            /// </summary>
            /// <returns></returns>
            public override Task ProcessAsync(object msg)
            {
                try
                {
    				// todo
                }
                catch (Exception e)
                {
                }
                return Task.CompletedTask;
            }
        }

大致是这样的，具体的案例，可以自己写一个