在不同任务中同时使用 ClientWebSocket、SendAsync 和 ReceiveAsync 方法的意外行为 （.Net Framework）

根据您告知的内容，问题是：

在 ReceiveTask 中，信号量中没有命令 to 只有 .为了使用信号量，所有任务都必须请求输入并通知发布。EnterRelease

当你这样做时，它会起作用，因为它给了时间。await Task.Delay(1)ReceiveTaskRelease

下面是一个模拟网络和操作的示例：SendAsyncReceiveAsync

using System.Text;

class q77091244
{
    static void Main(string[] args)
    {
        //Channel commandQueue = new Channel("command");       
        Channel responseQueue = new Channel();  

        // SendTask
        Task SendTask = Task.Run(async () => 
        {
            Console.WriteLine("SendTask started...");
            for (int i = 0; i < 10; i++)
            {
                CancellationToken cts = new CancellationToken();
                Console.WriteLine("SendTask Waiting");
                var command = await responseQueue.ReadAsync(cts);
                Console.WriteLine("SendTask Sending");
                await Task.Delay(100); //simulates await webSocket.SendAsync...
                //in order to log command sent, push it into response queue.
                Console.WriteLine("SendTask Push");
                responseQueue.Push(command);
            }
        });

        Task ReceiveTask = Task.Run(async () => 
        {
            Console.WriteLine("ReceiveTask started...");
            StringBuilder strBuilder = new StringBuilder();
            var buffer = Encoding.UTF8.GetBytes("item x");

            for (int i = 0; i < 10; i++)
            {
                CancellationToken cts = new CancellationToken();
                Console.WriteLine("ReceiveTask Waiting");
                await responseQueue.WaitAsync(cts);                
                // receive message
                strBuilder.Append(Encoding.UTF8.GetString(buffer, 0, buffer.Length));
                Console.WriteLine("ReceiveTask Receiving");
                await Task.Delay(150); //simultates await webSocket.ReceiveAsync(buffer, recCts);
                Console.WriteLine("ReceiveTask Push");
                responseQueue.Push(strBuilder.ToString());
                strBuilder.Clear();
            }
        });

        Thread.Sleep(10000);

    }
}

通道是相同的，但初始化为 1 以绿色状态启动（更多信息）且不操作：SemaphoreSlimWaitAsyncDequeue

using System.Collections.Concurrent;

public class Channel
{
        private ConcurrentQueue<string> _queue;
        private SemaphoreSlim _flag;

        public Channel()
        {
            _queue = new ConcurrentQueue<string>();
            _flag = new SemaphoreSlim(1);
        }
        
        public async Task WaitAsync(CancellationToken readCancelToken)
        {
            await _flag.WaitAsync(readCancelToken);
        }

        public async Task<string> ReadAsync(CancellationToken readCancelToken)
        {          
            await _flag.WaitAsync(readCancelToken);
            if (readCancelToken.IsCancellationRequested)
            {
                return default;
            }

            string item;
            if (_queue.TryDequeue(out item))
            {
                Console.WriteLine("dequed " + item);
                return item;
            }

            return default;
        }

        public void Push(string item)
        {
            Console.WriteLine("enqueue " + item);
            _queue.Enqueue(item);
            _flag.Release();
        }
    }

输出：

SendTask started...
SendTask Waiting
ReceiveTask started...
SendTask Sending
ReceiveTask Waiting
SendTask Push
enqueue 
SendTask Waiting
ReceiveTask Receiving
ReceiveTask Push
enqueue item x
ReceiveTask Waiting
dequed
SendTask Sending
SendTask Push
enqueue 
SendTask Waiting
ReceiveTask Receiving
ReceiveTask Push
enqueue item x
ReceiveTask Waiting
dequed item x
SendTask Sending
SendTask Push
enqueue item x
SendTask Waiting
ReceiveTask Receiving
ReceiveTask Push
enqueue item x
ReceiveTask Waiting
dequed
SendTask Sending
SendTask Push
enqueue