将使用事件循环的 C++/Qt 线程转换为使用 Dispatchers 的 C#（或 java）线程的可能性

Posted 2023-02-17

【中文标题】将使用事件循环的 C++/Qt 线程转换为使用 Dispatchers 的 C#（或 java）线程的可能性

【英文标题】：Possibility of converting a C++/Qt threads using event loops to C# (or java) threads with Dispatchers

【发布时间】：2021-05-06 13:09:41

【问题描述】：

是否可以转换使用线程执行事件循环并使用信号/槽机制和排队连接的 Qt/C++ 代码在 2 个线程之间进行通信？

我看到C#/WPF（或WinForms）中有一个叫Dispatcher的类，我不知道它是否存在于UI上下文之外（可以在.NET Console App或Core App中使用） ?)。

在 Qt 中，我通常使用这种模式：我有一个类 Application 来保存主窗口/小部件，它会启动一个线程来处理非 UI 内容（从 DAQ 卡获取、与其他设备通信等... )

Application::Application(QObject *parent) :
    QObject(parent)

s_instance = this;
//....

// Register types so they can be exchanged via the signal/slot mechanism
    qRegisterMetaType<Data>("Data");

// Com. manager
    m_workerThread = new QThread(this);
    m_comManager = new CommunicationManager();


    QObject::connect(m_comManager, &CommunicationManager::dataRcvd,
                     this,         &Application::onDataRcvd);
    QObject::connect(m_comManager, &CommunicationManager::error,
                     this,         &Application::showError);
    QObject::connect(m_comManager, &CommunicationManager::progress,
                     this,         &Application::setProgress);

    m_comManager->moveToThread(m_workerThread );
    m_workerThread ->start();

//...

在工作对象中，在某些情况下，我有一个方法可以执行一个循环，如果 UI 请求可以中断该循环。

在 UI 线程中，我可以请求工作线程执行这个函数（如发送邮件）以停止 for 循环：

QMetaObject::invokeMethod(m_comManager , "setContinue",
                                  Qt::QueuedConnection, Q_ARG(bool, false));

在辅助线程中“生活”的对象中，我有这个已经在执行的公共槽（如 setContinue，请求是通过使用 invokeMethod 的邮件/信号发送的）：

void CommunicationManager::processRequest()


// the for loop that can be interrupted from UI thread safely
for (size_t i = 0; i < numberOfLoops && !error && !isCanceled(); ++i)
        //.... 

//...


bool CommunicationManager::isCanceled()

    auto const dispatcher = QThread::currentThread()->eventDispatcher();
    if (!dispatcher)
    
        return false;
    
    dispatcher->processEvents(QEventLoop::AllEvents);
    return !m_bContinue;


// The method that is requested (from UI) to be executed on the worker thread :
void CommunicationManager::setContinue(const bool continue)

    if (m_bContinue!= continue)
    
        m_bContinue = continue;
    

是否可以使用 C#（或即使是 Java，我也很好奇）。如果答案是否定的，那真的很可悲，因为这避免了我使用锁及其缺点。

【参考方案1】：

异步消息传递

是否可以转换使用线程执行事件循环并使用信号/槽机制和排队连接的 Qt/C++ 代码在 2 个线程之间进行通信？

您的问题范围很广，所以我会尽力为您指明正确的方向。让我们从信号/槽机制开始，来自Qt documentation：

信号和槽用于对象之间的通信。信号和槽机制是 Qt 的核心特性，可能也是与其他框架提供的特性最不同的部分。 Qt 的meta-object system 使信号和槽成为可能。

需要说，您必须迁移到 DotNet 的消息传递框架之一才能实现类似的功能。根据您当前的设置和要求：

设置连接：

QObject::connect(m_comManager, &CommunicationManager::dataRcvd,
                 this,         &Application::onDataRcvd);
QObject::connect(m_comManager, &CommunicationManager::error,
                 this,         &Application::showError);
QObject::connect(m_comManager, &CommunicationManager::progress,
                 this,         &Application::setProgress);

基于消息的交互：

QMetaObject::invokeMethod(m_comManager , "setContinue",
                                  Qt::QueuedConnection, Q_ARG(bool, false));

邮箱队列：

...这避免了我使用锁及其缺点。

您可能想查看 Actors 模型，例如 Akka.NET 提供的。

同步上下文

我看到C#/WPF（或WinForms）中有一个叫Dispatcher的类，我不知道它是否存在于UI上下文之外（可以在.NET Console App或Core App中使用） ?)。

可以，我们稍后会讨论，但我认为它不能很好地映射到您的用例。如您所知，支持SynchronizationContext 通常标记具有特殊能力的线程，例如GUI 应用程序中的主线程。多亏了这种上下文，async / await 模式可用于轻松卸载工作并将结果应用回主线程。尤其是两个线程之间所需通信的双向性质，比SynchronizationContext 提供的更高级。更好地了解内部运作可能会有所帮助。首先，一些上下文...

控制台同步

Console 类同步输出流的使用，因此您可以从多个线程对其进行写入。对于控制台应用程序，这意味着没有两个线程可以同时写入屏幕。由于没有必要，你会发现SynchronizationContext 没有在主线程上设置。主线程是唯一的前台线程，从 ThreadPool 检索工作线程（使用默认设置和自定义线程创建除外）。

在 GUI 应用程序中，主线程是带有SynchronizationContext 的线程。每次在主线程上创建Task/async 是await-ed，此SynchronizationContext 存储在Task 上，因此可以在Task 完成时检索它。如果捕获到的SynchronizationContext 为空，那么将由原来的TaskScheduler（通常是TaskScheduler.Default，即ThreadPool）安排继续。

同步执行

让我们从一个重要的观察开始。使用await 并不一定意味着我们正在处理并发。在下面的示例中，我们使用Thread.CurrentThread.ManagedThreadId 查询当前线程的id。

namespace ConsoleApp

    internal class Program
    
        public static async Task Main()
        
            Console.WriteLine(GetCurrentThreadId());
            await DemoAsync();
            Console.ReadKey();
        

        private static Task DemoAsync()
        
            Console.WriteLine(GetCurrentThreadId());
            return Task.CompletedTask;
        

        private static int GetCurrentThreadId() => Thread.CurrentThread.ManagedThreadId;
    

输出：

1
1

哪个是主线程的id。所以只要我们在等待的DemoAsync中没有await任何async方法所有代码同步运行。换句话说，我们永远不会离开我们开始的主线程。该代码通过DemoAsync 的签名中缺少async 关键字来强调这一点。

异步执行

让我们修改前面的示例来说明差异。

namespace ConsoleApp

    internal class Program
    
        public static async Task Main()
        
            Console.WriteLine(GetCurrentThreadId());
            await DemoAsync();
            Console.ReadKey();
        

        private static async Task DemoAsync()
        
            Console.WriteLine(GetCurrentThreadId());
            await Task.Yield();
            Console.WriteLine(GetCurrentThreadId());
        

        private static int GetCurrentThreadId() => Thread.CurrentThread.ManagedThreadId;
    

输出：

1
1
3

您的输出中的最后一个 id 可能会有所不同。 Task.Yield() 创建一个 Task，在等待时，它将控制权返回到创建它的上下文。但是，在控制台应用程序的情况下，Task.Yield() 的使用有些特殊。鉴于SynchronizationContext 默认情况下未设置，延续不知道将控制权返回到何处。因此，它保留在当前上下文中，即ThreadPool 的（使用的线程）的上下文中。

当然，我们可以等待任何async 方法来获得类似的结果，但它为以下部分提供了一个很好的介绍。

同步上下文

参考Await, SynchronizationContext, and Console Apps | .NET Parallel Programming (microsoft.com)

以下引用帖子中的示例演示了控制权是如何永远不会交还给主线程的。只有导致第一个 Task.Yield() 的代码在主线程上运行。从那时起，只使用来自ThreadPool 的线程。

namespace ConsoleApp

    internal class Program
    
        public static async Task Main()
        
            await DemoAsync();
            Console.ReadKey();
        

        private static async Task DemoAsync()
        
            var d = new Dictionary<int, int>();
            for (var i = 0; i < 10000; i++)
            
                var id = GetCurrentThreadId();
                d[id] = d.TryGetValue(id, out var count) ? count + 1 : 1;
                await Task.Yield();
            
            foreach (var pair in d) Console.WriteLine(pair);
        

        private static int GetCurrentThreadId() => Thread.CurrentThread.ManagedThreadId;
    

输出（你的会有所不同）：

[1, 1]
[3, 3087]
[4, 3292]
[5, 2667]
[6, 953]

为了让 yield 找到归宿，我们将实现一个自定义 SynchronizationContext。这不是您会发现自己经常做的事情，因为每个平台通常都提供自己的自定义实现。我们使用BlockingCollection，因为它是我们消息泵的一个很好的候选者。它不仅默认具有queue语义，而且在队列为空时也会阻塞调用者。

namespace ConsoleApp

    public sealed class SingleThreadSynchronizationContext : SynchronizationContext
    
        private readonly BlockingCollection<KeyValuePair<SendOrPostCallback, object>> _queue =
            new BlockingCollection<KeyValuePair<SendOrPostCallback, object>>();
        
        public override void Post(SendOrPostCallback d, object state) =>
            _queue.Add(new KeyValuePair<SendOrPostCallback, object>(d, state));

        public void RunOnCurrentThread()
        
            foreach (var workItem in _queue.GetConsumingEnumerable())
            
                workItem.Key(workItem.Value);
            
        

        public void Complete() => _queue.CompleteAdding();
    

接下来，在我们更新的示例中，我们首先存储当前上下文（同样，在控制台应用程序中将是 null），然后创建我们的自定义 SynchronizationContext 并将其设置为当前线程的上下文。 DemoAsync 的执行现在将在其迭代期间保留在主线程上。

namespace ConsoleApp

    internal class Program
    
        public static void Main()
        
            var prevCtx = SynchronizationContext.Current;

            try
            
                var syncCtx = new SingleThreadSynchronizationContext();
                SynchronizationContext.SetSynchronizationContext(syncCtx);

                var t = DemoAsync();
                t.ContinueWith(_ => syncCtx.Complete(), TaskScheduler.Default);

                syncCtx.RunOnCurrentThread();

                t.GetAwaiter().GetResult();
            

            finally
            
                SynchronizationContext.SetSynchronizationContext(prevCtx);
            

            Console.ReadKey();
        

        private static async Task DemoAsync()
        
            var d = new Dictionary<int, int>();
            for (var i = 0; i < 10000; i++)
            
                var id = GetCurrentThreadId();
                d[id] = d.TryGetValue(id, out var count) ? count + 1 : 1;
                await Task.Yield();
            
            foreach (var pair in d) Console.WriteLine(pair);
        

        private static int GetCurrentThreadId() => Thread.CurrentThread.ManagedThreadId;
    

输出：

[1, 10000]

结束

在 DotNet 中，通常不鼓励创建自定义线程，而是支持使用来自 ThreadPool 的线程。 CPU 密集型任务和 I/O 密集型任务都有专用线程，因此其中一个的重负载不会影响另一个。简而言之，您并不真正（需要）关心哪个线程处理您的请求。这并不意味着具有无锁方法的 Actor 模型没有它的好处。这不是原生的。

其他资源

It's All About the SynchronizationContext | The Implementations of SynchronizationContext (MSDN magazine)

DispatcherSynchronizationContext (WindowsBase.dll: System.Windows.Threading) WPF 和 Silverlight 应用程序使用 DispatcherSynchronizationContext，它将委托以“正常”优先级排列到 UI 线程的 Dispatcher .当线程通过调用 Dispatcher.Run 开始其 Dispatcher 循环时，此 SynchronizationContext 被安装为当前上下文。 DispatcherSynchronizationContext 的上下文是单个 UI 线程。

排队到 DispatcherSynchronizationContext 的所有委托由特定的 UI 线程按照它们排队的顺序一次执行一个。当前实现为每个***窗口创建一个 DispatcherSynchronizationContext，即使它们都共享相同的底层 Dispatcher。

默认 (ThreadPool) SynchronizationContext (mscorlib.dll: System.Threading) 默认 SynchronizationContext 是默认构造的 SynchronizationContext 对象。按照惯例，如果一个线程的当前 SynchronizationContext 为 null，那么它就隐含了一个默认的 SynchronizationContext。

默认的 SynchronizationContext 将其异步委托排队到 ThreadPool，但直接在调用线程上执行其同步委托。因此，它的上下文涵盖了所有 ThreadPool 线程以及任何调用 Send 的线程。上下文“借用”调用 Send 的线程，将它们带入其上下文，直到委托完成。从这个意义上说，默认上下文可能包括进程中的any线程。

除非代码由 ASP.NET 托管，否则默认 SynchronizationContext 将应用于 ThreadPool 线程。默认 SynchronizationContext 也隐式应用于显式子线程（Thread 类的实例），除非子线程设置自己的 SynchronizationContext。因此，UI 应用程序通常有两个同步上下文：覆盖 UI 线程的 UI SynchronizationContext 和覆盖 ThreadPool 线程的默认 SynchronizationContext。

Messages and Agents | F# for fun and profit (fsharpforfunandprofit.com)

【参考方案2】：

通过在 C++ 应用程序中嵌入 JVM，您可以使用 Java 反射在辅助线程中调用任何 Java 方法。

见：this link

但一般来说，低级实现（在 C、C++ 等中）是从更高级别调用的，而不是反向调用。

【讨论】：

我想知道是否可以在本地执行此操作（使用 C# 和/或 Java 中的线程的消息循环）。显然答案是否定的。在 C# 中，也有异步方法、取消标记和可能的事件，但不是我可以在 Qt 中使用的那种模式。