实现线程池的调度器

Unity/Assets/Scripts/Core/Game/Game.cs

@@ -47,6 +47,10 @@ namespace ET
         private readonly Dictionary<int, Process> processes = new();
 
         private int idGenerator;
+
+        private Thread thread;
+
+        private bool isStart;
         
         public Process Create()
         {
@@ -70,13 +74,21 @@ namespace ET
             }
         }
 
-        public void Loop()
+        public void Start()
         {
-            lock (this)
+            this.thread = new Thread(() =>
             {
-                this.Update();
-                this.LateUpdate();
-            }
+                while (this.isStart)
+                {
+                    lock (this)
+                    {
+                        this.Update();
+                        this.LateUpdate();
+                    }
+                    Thread.Sleep(1);
+                }
+            });
+            this.thread.Start();
         }
         
         public T AddSingleton<T>() where T: Singleton<T>, new()
@@ -317,6 +329,9 @@ namespace ET
 
         public void Dispose()
         {
+            this.isStart = false;
+            this.thread.Join();
+            
             using (Locker _ = new())
             {
                 // 顺序反过来清理

Unity/Assets/Scripts/Core/Game/ISingletonScheduler.cs

@@ -7,7 +7,5 @@
         void StopScheduler();
 
         void Add(Process process);
-        
-        void Remove(Process process);
     }
 }

Unity/Assets/Scripts/Core/Game/Module/ProcessActor.meta

@@ -1,8 +0,0 @@
-fileFormatVersion: 2
-guid: b4462a51f8c36c84a90d48e6315b66b4
-folderAsset: yes
-DefaultImporter:
-  externalObjects: {}
-  userData: 
-  assetBundleName: 
-  assetBundleVariant: 

Unity/Assets/Scripts/Core/Game/Module/ProcessScheduler/ThreadPoolScheduler.cs

@@ -1,79 +1,98 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Threading;
 
 namespace ET
 {
-    public class ThreadPoolScheduler: Singleton<ThreadPoolScheduler>, ISingletonScheduler, ISingletonUpdate
+    public class ThreadPoolScheduler: Singleton<ThreadPoolScheduler>, ISingletonScheduler
     {
         private bool isStart;
-        private Dictionary<int, Process> Processes { get; } = new();
-        private readonly ThreadSynchronizationContext threadSynchronizationContext = new();
+        
+        private readonly HashSet<Thread> threads = new();
+        
+        public int ThreadCount { get; set; }
+
+        private readonly ConcurrentDictionary<int, Process> dictionary = new();
+
+        private readonly ConcurrentQueue<int> idQueue = new();
 
         public void StartScheduler()
         {
             this.isStart = true;
-        }
-        
-        public void Update()
-        {
-            if (!this.isStart)
+            for (int i = 0; i < this.ThreadCount; ++i)
             {
-                return;
+                this.threads.Add(new Thread(this.Loop));
             }
-            
-            this.threadSynchronizationContext.Update();
-            
-            foreach ((int _, Process process) in this.Processes)
-            {
-                if (process.IsRuning)
-                {
-                    continue;
-                }
 
-                process.IsRuning = true;
-                ThreadPool.QueueUserWorkItem(process.Loop);
+            foreach (Thread thread in this.threads)
+            {
+                thread.Start();
             }
         }
 
-        public void StopScheduler()
+        private void Loop()
         {
-            this.isStart = false;
-            
-            // 等待线程池中的Process Loop完成
-            while (true)
+            while (this.isStart)
             {
-                int count = 0;
-                
-                foreach ((int _, Process process) in this.Processes)
+                try
                 {
+                    if (!this.idQueue.TryDequeue(out int id))
+                    {
+                        Thread.Sleep(1);
+                        continue;
+                    }
+
+                    if (!this.dictionary.TryGetValue(id, out Process process))
+                    {
+                        Thread.Sleep(1);
+                        continue;
+                    }
+
+                    // 执行过的或者正在执行的进程放到队尾
                     if (process.IsRuning)
                     {
-                        break;
+                        process.LoopOnce();
                     }
 
-                    ++count;
+                    this.idQueue.Enqueue(id);
+                    
+                    Thread.Sleep(1);
                 }
-
-                if (count == this.Processes.Count)
+                catch (Exception e)
                 {
-                    break;
+                    Log.Error(e);
                 }
             }
         }
 
-        public void Add(Process process)
+        public void StopScheduler()
         {
-            lock (Game.Instance)
+            this.isStart = false;
+            foreach (Thread thread in this.threads)
             {
-                this.Processes.Add(process.Id, process);
+                thread.Join();
             }
         }
 
-        public void Remove(Process process)
+        public void Add(Process process)
         {
+            int id = 0;
             lock (Game.Instance)
             {
-                this.Processes.Remove(process.Id);
+                id = process.Id;
+                if (id == 0)
+                {
+                    return;
+                }
+                
+                if (this.dictionary.ContainsKey(id))
+                {
+                    return;
+                }
+
+                this.dictionary[id] = process;
+                this.idQueue.Enqueue(id);
             }
         }
     }

Unity/Assets/Scripts/Core/Module.meta

@@ -1,8 +0,0 @@
-fileFormatVersion: 2
-guid: fd94f671f44cc5449b58ec76f5740fb7
-folderAsset: yes
-DefaultImporter:
-  externalObjects: {}
-  userData: 
-  assetBundleName: 
-  assetBundleVariant: 

Unity/Assets/Scripts/Core/Process/Process.cs

@@ -15,15 +15,6 @@ namespace ET
         public Process(int id)
         {
             this.Id = id;
-
-            this.loop = (_) =>
-            {
-                this.Register();
-                this.Update();
-                this.LateUpdate();
-                this.FrameFinishUpdate();
-                this.IsRuning = false;
-            };
         }
 
         private readonly Stack<IProcessSingleton> singletons = new();
@@ -35,8 +26,6 @@ namespace ET
         private readonly Queue<IProcessSingleton> loads = new();
 
         private readonly Queue<ETTask> frameFinishTask = new();
-        
-        private readonly WaitCallback loop;
 
         private void Register()
         {
@@ -149,12 +138,12 @@ namespace ET
             }
         }
 
-        public WaitCallback Loop
+        public void LoopOnce()
         {
-            get
-            {
-                return this.loop;
-            }
+            this.Register();
+            this.Update();
+            this.LateUpdate();
+            this.FrameFinishUpdate();
         }
         
         public void Load()
@@ -207,8 +196,6 @@ namespace ET
             }
             
             this.Id = 0;
-            
-            Game.Instance.Remove(id);
 
             this.IsRuning = false;
             

Unity/Assets/Scripts/Core/Singleton.meta

@@ -1,8 +0,0 @@
-fileFormatVersion: 2
-guid: 02e6bdc5be01d6548adf28e72cb73a3f
-folderAsset: yes
-DefaultImporter:
-  externalObjects: {}
-  userData: 
-  assetBundleName: 
-  assetBundleVariant: 

Unity/Assets/Scripts/Core/World.meta

@@ -1,8 +0,0 @@
-fileFormatVersion: 2
-guid: 780f97adbed71ad4fa77ce29bb21f6be
-folderAsset: yes
-DefaultImporter:
-  externalObjects: {}
-  userData: 
-  assetBundleName: 
-  assetBundleVariant: 

Unity/Assets/Scripts/Loader/UnityScheduler.cs

@@ -10,7 +10,7 @@ namespace ET
 
         public void StartScheduler()
         {
-            Init.Instance.ThreadSynchronizationContext.Post(() => { Init.Instance.IsStart = true; });
+            Init.Instance.IsStart = true;
         }
 
         public void StopScheduler()
@@ -29,14 +29,11 @@ namespace ET
 
         public void Add(Process process)
         {
-            this.process = process;
-            Init.Instance.ThreadSynchronizationContext.Post(()=>{ Init.Instance.Process = process; });
-        }
-        
-        public void Remove(Process process)
-        {
-            this.process = null;
-            Init.Instance.ThreadSynchronizationContext.Post(()=>{ Init.Instance.Process = null; });
+            lock (Game.Instance)
+            {
+                this.process = process;
+                Init.Instance.ThreadSynchronizationContext.Post(() => { Init.Instance.Process = process; });
+            }
         }
     }
 }