List.Add没有完美地被并行和异步添加

Question

我有一个Blazor应用程序，它需要每秒钟调用一个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)。但是，如果我在Console.WriteLine之前添加emptyTestlist.Add(i)，它就能正常工作(emptyTestlist.Count总是4)。

我不知道怎么解决。有什么问题吗？

Answer 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或推送通知将任何新数据通知客户端。

借用文档示例，这就足够从服务器接收消息了：

@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();
    }