使用 SemaphoreSlim 和 Dictionary 等待 NetMQ 回复

Question

我有两个应用程序使用 NetMQ 相互通信。第一个应用程序内置了一个 API 控制器，第二个应用程序（引擎）在必要时与第一个应用程序来回通信。所有 API 控制器函数都是异步任务，因为我们期望同时发生许多请求。其中一个 API 控制器函数需要首先向第二个应用程序（引擎）发送 NetMQ 消息，并等待其响应，然后再将结果返回给 API 函数请求者（API 用户）。它需要以异步方式完成所有这些，而不需要从线程池中获取，因为正如我之前所说，我们可能有大量请求不断地访问此控制器。当 SemaphoreSlim 等待时，第二个应用程序（引擎）正在更新数据库中特定记录的状态，并且我需要在 API 控制器函数中读取该记录的状态并将其返回给 API 用户，如果我没有 SemaphoreSlim 等待设计，那么代码当然会提前读取数据库记录并且不会返回正确的状态。因此，解决方案是创建等待，允许引擎更新状态，然后通知 API 控制器函数它可以从数据库读取新更新的状态。一位同事告诉我，使用 SemaphoreSlim 和 Dictionary 设置通过锁定在函数中创建等待，并且仅在收到来自引擎的 NetMQ 回复时才从字典中释放 SemaphoreSlim。然后代码自然会继续，我将检查数据库的更新状态并将其返回给 API 用户。问题在于 API 控制器函数完全忽略了 SemaphoreSlim 锁并过早地从数据库读取。最有可能的是因为它是异步的，所以我的问题是，如何使 SempahoreSlim 锁定和读取数据库代码以异步方式一起工作，而不锁定线程？我相信代码会让您更好地理解我正在尝试做的事情。1. API控制器部分功能代码：//Call EngineMessenging.Queue.Enqueue(OrderPositionText + " " + UserId + " " + AssetPairId + " " + OrderType + " " + OrderRequestId + " " + rAmount + " " + rPrice + " " + Stop);//Create SemaphoreSlim Lock and add it to global Dictionary     var NewSemaphore = new SemaphoreSlim(1, 1);await NewSemaphore.WaitAsync().ConfigureAwait(false);ApiHub.UserSemaphoreDictionary.TryAdd(UserId, NewSemaphore);//Lock was released lets now read record in Database then return appropriate result to API user.                Debug.WriteLine("Checking order status for " + uuid);                OrderRequest OrderRequestRec = await _context.OrderRequest.Where(x => x.UUID == uuid).FirstOrDefaultAsync();    if (OrderRequestRec != null)                    {                        Debug.WriteLine("ORDER STATUS: " + OrderRequestRec.Status);    }2.这是另一个类中接收NetMQ消息并释放SemaphoreSlim的代码：                        //API Semaphore Release lock                        SemaphoreSlim checkUserId;                        if (ApiHub.UserSemaphoreDictionary.TryGetValue(UserId, out checkUserId)) {                            Debug.WriteLine("RELEASING LOCK: " + UserId);                            checkUserId.Release();                        }结果是先读取Database，然后释放SemaphoreSlim。我需要释放 SemaphoreSlim，然后读取数据库。

Answer 1

对于这个场景，我使用 aTaskCompletionSource<bool>作为一种高级事件。它具有Task可直接等待的属性。然后，您可以从接收消息的类中调用SetResult(true)来完成任务并允许等待者继续。

请注意，类型bool是无关紧要的：任务必须具有某种结果类型，但在这里我们只是将其用作表示事件已发生的信号方式（也就是说，我们实际上只想要 a 等待，而不是Taska Task<T>） 。

编辑以响应OP的编辑：

是的，这是完全异步的。该await关键字使 C# 编译器将控制器的代码分解为多个部分，因此后面的部分成为await可以稍后恢复的“延续”。它将延续添加到任务的完成处理程序列表中，然后放弃线程；控制器方法不再执行，因此不会耗尽线程/堆栈。

在您调用的类中SetResult，这实际上会导致任务通知（执行）处理程序，从而导致控制器的代码唤醒并完成。无需创建额外的线程；现有的被使用，并且一旦没有立即要做的工作就被放弃。