快捷導(dǎo)航

C# 并發(fā)控制框架之單線程環(huán)境下實(shí)現(xiàn)每秒百萬級調(diào)度

更新時(shí)間：2024年10月17日 09:46:53 作者：小碼編匠

本文介紹了一款專為工業(yè)自動(dòng)化及機(jī)器視覺開發(fā)的C#并發(fā)流程控制框架,通過模仿Go語言并發(fā)模式設(shè)計(jì),支持高頻調(diào)度及復(fù)雜任務(wù)處理,已在多個(gè)項(xiàng)目中驗(yàn)證其穩(wěn)定性和可靠性

前言

在工業(yè)自動(dòng)化和機(jī)器視覺領(lǐng)域，對實(shí)時(shí)性、可靠性和效率的要求越來越高。為了滿足這些需求，我們開發(fā)了一款專為工業(yè)自動(dòng)化運(yùn)動(dòng)控制和機(jī)器視覺流程開發(fā)設(shè)計(jì)的 C# 并發(fā)流程控制框架。

該框架不僅適用于各種工業(yè)自動(dòng)化場景，還能在單線程環(huán)境下實(shí)現(xiàn)每秒百萬次以上的調(diào)度頻率，從而從容應(yīng)對涉及成千上萬輸入輸出點(diǎn)數(shù)的復(fù)雜任務(wù)。

并發(fā)流程控制框架

本框架提供一種全新的并發(fā)流程控制框架，它借鑒了Golang語言中的高效并發(fā)模式，并在此基礎(chǔ)上進(jìn)行了必要的功能擴(kuò)展。框架不僅能夠支持自定義的單/多線程調(diào)度機(jī)制，還允許在主UI線程中進(jìn)行調(diào)度，從而簡化了邏輯與用戶界面之間的交互。

另外，該框架還集成了高精度定時(shí)器、可配置的調(diào)度優(yōu)先級、邏輯停止與暫停等功能，讓我們能夠更加靈活地管理和控制復(fù)雜的自動(dòng)化流程。

框架優(yōu)勢

相較于傳統(tǒng)模型：相對于傳統(tǒng)的多線程模型、狀態(tài)機(jī)模型以及類PLC模型，本框架具有更加緊湊清晰的邏輯結(jié)構(gòu)，顯著提升了開發(fā)效率，并簡化了后續(xù)的維護(hù)與升級過程。
受Go語言啟發(fā)：框架的設(shè)計(jì)借鑒了 Go 語言中的高效并發(fā)模式，并在此基礎(chǔ)上進(jìn)行了必要的功能擴(kuò)展，以適應(yīng)工業(yè)自動(dòng)化領(lǐng)域的具體需求。
靈活的調(diào)度機(jī)制：支持自定義單線程或多線程調(diào)度，同時(shí)也可在主 UI 線程中進(jìn)行調(diào)度，便于邏輯與用戶界面的交互，增強(qiáng)了用戶體驗(yàn)。
豐富的內(nèi)置功能：內(nèi)置高精度定時(shí)器、可配置的調(diào)度優(yōu)先級、邏輯停止與邏輯暫停功能，確保任務(wù)執(zhí)行的準(zhǔn)確性和可控性。
樹形多任務(wù)調(diào)度：采用樹形結(jié)構(gòu)管理多任務(wù)調(diào)度，提高了邏輯的可靠性和系統(tǒng)的整體穩(wěn)定性。
卓越的性能表現(xiàn)：在單線程環(huán)境下，每秒可實(shí)現(xiàn)超過一百萬次的調(diào)度頻率，能夠從容應(yīng)對成千上萬輸入輸出點(diǎn)數(shù)的復(fù)雜場景。
廣泛的實(shí)踐驗(yàn)證：該框架已在多個(gè)實(shí)際項(xiàng)目中成功應(yīng)用，證明了其穩(wěn)定性和可靠性。

框架示例

代碼中定義了一系列不同的任務(wù)執(zhí)行模式，展示如何通過不同的調(diào)度策略來管理并發(fā)任務(wù)。

全局變量

static shared_strand strand：全局共享的調(diào)度器，用于保證線程安全。

日志記錄函數(shù)

Log(string msg)：記錄帶有時(shí)間戳的日志信息到控制臺。

工作任務(wù)函數(shù)

Worker(string name, int time = 1000)：模擬一個(gè)簡單的任務(wù)，該任務(wù)會(huì)在指定的毫秒數(shù)后打印一條消息。

主函數(shù)

MainWorker()：異步主任務(wù)函數(shù)，依次調(diào)用前面定義的各種任務(wù)模式。

Main(string[] args)：程序入口點(diǎn)，初始化工作服務(wù)、共享調(diào)度器，并啟動(dòng)主任務(wù)。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Go;
namespace WorkerFlow
{
    class Program
    {
        static shared_strand strand;
        static void Log(string msg)
        {
            Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")} {msg}");
        }
        static async Task Worker(string name, int time = 1000)
        {
            await generator.sleep(time);
            Log(name);
        }
        //1 A、B、C依次串行
        //A->B->C
        static async Task Worker1()
        {
            await Worker("A");
            await Worker("B");
            await Worker("C");
        }
        //2 A、B、C全部并行，且依賴同一個(gè)strand(隱含參數(shù)，所有依賴同一個(gè)strand的任務(wù)都是線程安全的)
        //A
        //B
        //C
        static async Task Worker2()
        {
            generator.children children = new generator.children();
            children.go(() => Worker("A"));
            children.go(() => Worker("B"));
            children.go(() => Worker("C"));
            await children.wait_all();
        }
        //3 A執(zhí)行完后，B、C再并行
        //  -->B
        //  |
        //A->
        //  |
        //  -->C
        static async Task Worker3()
        {
            await Worker("A");
            generator.children children = new generator.children();
            children.go(() => Worker("B"));
            children.go(() => Worker("C"));
            await children.wait_all();
        }
        //4 B、C都并行執(zhí)行完后，再執(zhí)行A
        //B--
        //  |
        //  -->A
        //  |
        //C--
        static async Task Worker4()
        {
            generator.children children = new generator.children();
            children.go(() => Worker("B"));
            children.go(() => Worker("C"));
            await children.wait_all();
            await Worker("A");
        }
        //5 B、C任意一個(gè)執(zhí)行完后，再執(zhí)行A
        //B--
        //  |
        //  >-->A
        //  |
        //C--
        static async Task Worker5()
        {
            generator.children children = new generator.children();
            var B = children.tgo(() => Worker("B", 1000));
            var C = children.tgo(() => Worker("C", 2000));
            var task = await children.wait_any();
            if (task == B)
            {
                Log("B成功");
            }
            else
            {
                Log("C成功");
            }
            await Worker("A");
        }
        //6 等待一個(gè)特定任務(wù)
        static async Task Worker6()
        {
            generator.children children = new generator.children();
            var A = children.tgo(() => Worker("A"));
            var B = children.tgo(() => Worker("B"));
            await children.wait(A);
        }
        //7 超時(shí)等待一個(gè)特定任務(wù)，然后中止所有任務(wù)
        static async Task Worker7()
        {
            generator.children children = new generator.children();
            var A = children.tgo(() => Worker("A", 1000));
            var B = children.tgo(() => Worker("B", 2000));
            if (await children.timed_wait(1500, A))
            {
                Log("成功");
            }
            else
            {
                Log("超時(shí)");
            }
            await children.stop();
        }
        //8 超時(shí)等待一組任務(wù)，然后中止所有任務(wù)
        static async Task Worker8()
        {
            generator.children children = new generator.children();
            children.go(() => Worker("A", 1000));
            children.go(() => Worker("B", 2000));
            var tasks = await children.timed_wait_all(1500);
            await children.stop();
            Log($"成功{tasks.Count}個(gè)");
        }
        //9 超時(shí)等待一組任務(wù)，然后中止所有任務(wù)，且在中止任務(wù)中就地善后處理
        static async Task Worker9()
        {
            generator.children children = new generator.children();
            children.go(() => Worker("A", 1000));
            children.go(async delegate ()
            {
                try
                {
                    await Worker("B", 2000);
                }
                catch (generator.stop_exception)
                {
                    Log("B被中止");
                    await generator.sleep(500);
                    throw;
                }
                catch (System.Exception)
                {
                }
            });
            var task = await children.timed_wait_all(1500);
            await children.stop();
            Log($"成功{task.Count}個(gè)");
        }
        //10 嵌套任務(wù)
        static async Task Worker10()
        {
            generator.children children = new generator.children();
            children.go(async delegate ()
            {
                generator.children children1 = new generator.children();
                children1.go(() => Worker("A"));
                children1.go(() => Worker("B"));
                await children1.wait_all();
            });
            children.go(async delegate ()
            {
                generator.children children1 = new generator.children();
                children1.go(() => Worker("C"));
                children1.go(() => Worker("D"));
                await children1.wait_all();
            });
            await children.wait_all();
        }
        //11 嵌套中止
        static async Task Worker11()
        {
            generator.children children = new generator.children();
            children.go(() => Worker("A", 1000));
            children.go(async delegate ()
            {
                try
                {
                    generator.children children1 = new generator.children();
                    children1.go(async delegate ()
                    {
                        try
                        {
                            await Worker("B", 2000);
                        }
                        catch (generator.stop_exception)
                        {
                            Log("B被中止1");
                            await generator.sleep(500);
                            throw;
                        }
                        catch (System.Exception)
                        {
                        }
                    });
                    await children1.wait_all();
                }
                catch (generator.stop_exception)
                {
                    Log("B被中止2");
                    throw;
                }
                catch (System.Exception)
                {
                }
            });
            await generator.sleep(1500);
            await children.stop();
        }
        //12 并行執(zhí)行且等待一組耗時(shí)算法
        static async Task Worker12()
        {
            wait_group wg = new wait_group();
            for (int i = 0; i < 2; i++)
            {
                wg.add();
                int idx = i;
                var _ = Task.Run(delegate ()
                {
                    try
                    {
                        Log($"執(zhí)行算法{idx}");
                    }
                    finally
                    {
                        wg.done();
                    }
                });
            }
            await wg.wait();
            Log("執(zhí)行算法完成");
        }
        //13 串行執(zhí)行耗時(shí)算法，耗時(shí)算法必需放在線程池中執(zhí)行，否則依賴同一個(gè)strand的調(diào)度將不能及時(shí)
        static async Task Worker13()
        {
            for (int i = 0; i < 2; i++)
            {
                await generator.send_task(() => Log($"執(zhí)行算法{i}"));
            }
        }
        static async Task MainWorker()
        {
            await Worker1();
            await Worker2();
            await Worker3();
            await Worker4();
            await Worker5();
            await Worker6();
            await Worker7();
            await Worker8();
            await Worker9();
            await Worker10();
            await Worker11();
            await Worker12();
            await Worker13();
        }
        static void Main(string[] args)
        {
            work_service work = new work_service();
            strand = new work_strand(work);
            generator.go(strand, MainWorker);
            work.run();
            Console.ReadKey();
        }
    }
}

框架地址

Gitee： https://gitee.com/hamasm/CsGo

總結(jié)

值得一提的是，該框架特別設(shè)計(jì)用于工業(yè)自動(dòng)化運(yùn)動(dòng)控制以及機(jī)器視覺流程開發(fā)領(lǐng)域，其獨(dú)特的樹形多任務(wù)調(diào)度機(jī)制極大提高了邏輯的可靠性，同時(shí)單線程環(huán)境下的每秒調(diào)度次數(shù)可達(dá)一百萬次以上，足以應(yīng)對涉及成千上萬輸入輸出點(diǎn)數(shù)的應(yīng)用場景。經(jīng)過多個(gè)項(xiàng)目的實(shí)際驗(yàn)證，證明了其穩(wěn)定性和可靠性，為工業(yè)自動(dòng)化提供了強(qiáng)有力的支持。

通過本文的介紹，希望能為工業(yè)自動(dòng)化領(lǐng)域的開發(fā)者提供一個(gè)高效、可靠且易于使用的工具。借助這一工具，大家在構(gòu)建復(fù)雜的控制系統(tǒng)時(shí)，能夠更加輕松地應(yīng)對并發(fā)處理的挑戰(zhàn)。也期待您在評論區(qū)留言交流，分享您的寶貴經(jīng)驗(yàn)和建議。

到此這篇關(guān)于C# 并發(fā)控制框架之單線程環(huán)境下實(shí)現(xiàn)每秒百萬級調(diào)度的文章就介紹到這了,更多相關(guān)C# 每秒百萬級調(diào)度內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: