List.Add 没有与 Parallel 和 async 完美添加

Posted 2023-03-31

【中文标题】List.Add 没有与 Parallel 和 async 完美添加

【英文标题】：List.Add not added perfectly with Parallel and async

【发布时间】：2021-10-13 06:58:07

【问题描述】：

我有一个 Blazor WASM 应用程序，它需要每秒调用一次 API 而不会阻塞 UI。这段代码演示了我是如何尝试这样做的：

List<int> testList = new();
testList.Add(1);
testList.Add(2);
testList.Add(3);
testList.Add(4);

List<int> emptyTestlist = new();

CancellationTokenSource cts;

Test();

void Test()

    Parallel.Invoke(async () =>
    
        do
        
            Console.WriteLine("Start");
            await Task.Delay(1000);
            await Test2();
            Console.WriteLine("END");
         while (true);
    );

Console.ReadLine();

async ValueTask Test2()

    emptyTestlist.Clear();
    cts = new();
    await Parallel.ForEachAsync(testList, cts.Token, async (test, token) =>
    
        await Test4(test);
    );
    foreach (var test in emptyTestlist)
    
        await Test3(test);
    


async Task Test4(int i)

    await Task.Delay(300);
    //Console.WriteLine("if I Add this console.WriteLine It's added perfectly");
    emptyTestlist.Add(i);
    Console.WriteLine($"from TEST4: i");


async Task Test3(int i)

    Console.WriteLine($"TEST3 i.");
    await Task.Delay(1000);
    Console.WriteLine($"TEST3 i, after 1sec");

如果我评论Console.WriteLine("if I Add this console.WriteLine It's added perfectly"); 行，它的添加并不完美。 （emptyTestlist.Count 并不总是 4）。但如果我在emptyTestlist.Add(i) 之前添加Console.WriteLine，它就可以正常工作（emptyTestlist.Count 始终为 4）。

我不知道如何解决它。有什么问题？

【问题讨论】：

“有什么问题。” - 您正在使用来自多个线程的非线程安全集合。任何事情都有可能发生。

@otterotter 你一开始想做什么？此代码使用了不合适的类和方法。 Parallel.Invoke 的使用就像是 Task.Run 对于初学者或更糟的是 Thread.Start。任务不是线程，它们使用线程。为什么不使用普通的旧计时器？

@otterotter 要避免线程问题，您需要使用锁或线程安全集合，如 ConcurrentQueue 或 ConcurrentDictionary。没有并发列表。如果你想要一个发布/订阅集合，你可以使用 Channel。

这些小伪示例的问题在于，它并不代表您要解决的实际问题，因此除了说，您需要做此线程安全，带有锁或线程安全集合 docs.microsoft.com/en-us/dotnet/standard/collections/… 或 docs.microsoft.com/en-us/dotnet/csharp/language-reference/…

见X/Y 问题。描述你正在努力实现的实际目标。还可以找到一些关于线程安全的资源，因为使用不安全的集合只是多线程程序的许多可能危害之一。

【参考方案1】：

轮询 API 的最简单方法是使用计时器：

@code 
    private List<Customer> custs=new List<Customer>();
    
    private System.Threading.Timer timer;

    protected override async Task OnInitializedAsync()
    
        await base.OnInitializedAsync();
        custs = await Http.GetFromJsonAsync<List<Customer>>(url);

        timer = new System.Threading.Timer(async _ =>
        
            custs = await Http.GetFromJsonAsync<List<Customer>>("/api/customers");
            InvokeAsync(StateHasChanged); 
        , null, 1000, 1000);
    

在这种情况下需要InvokeAsync(StateHasChanged);，因为状态是从计时器线程修改的，而 Blazor 不知道数据已更改。

如果我们想将结果添加到列表中，我们要么必须使用锁，要么必须使用线程安全集合，例如 ConcurrentQueue。

@code 
    private ConcurrentQueue<Customer> custs=new ConcurrentQueue<Customer>();
    
    private System.Threading.Timer timer;

    protected override async Task OnInitializedAsync()
    
        await base.OnInitializedAsync();
        custs = await Http.GetFromJsonAsync<List<Customer>>(url);

        timer = new System.Threading.Timer(async _ =>
        
            var results = await Http.GetFromJsonAsync<List<Customer>>("/api/customers");
            foreach(var c in results)
            
                custs.Enqueue(c);
            
            InvokeAsync(StateHasChanged); 
        , null, 1000, 1000);
    

每秒轮询一个 API 以防万一有任何新数据，但效率不是很高。最好让 API 使用 SignalR 或 Push Notifications 通知客户端任何新数据

从documentation example借用这足以接收来自服务器的消息：

@code 
    private HubConnection hubConnection;
    private List<string> messages = new List<string>();
    private string userInput;
    private string messageInput;

    protected override async Task OnInitializedAsync()
    
        hubConnection = new HubConnectionBuilder()
            .WithUrl(NavigationManager.ToAbsoluteUri("/chathub"))
            .Build();

        hubConnection.On<string, string>("ReceiveMessage", (user, message) =>
        
            var encodedMsg = $"user: message";
            messages.Add(encodedMsg);
            StateHasChanged();
        );

        await hubConnection.StartAsync();
    

【讨论】：

谢谢。他们只为我提供 api。所以不能使用集线器。谢谢！