Перейти к содержимому
Редактировал(а) Alexandr Fokin 2023/01/23 23:00
Последние авторы
1 [[Асинхронность. Async Await>>doc:Разработка.NET.C#.Многопоточность и асинхронность.Асинхронность\. Async Await.WebHome]]
2
3 [[Модели>>doc:Архитектура и модели.Модели.WebHome]]
4
5 ----
6
7 Используемые инструменты
8 1) Пакетная обработка
9 2) LongPooling
10
11
12 |(% style="width:105px" %)Исполнитель|(% style="width:1370px" %)Алгоритм
13 |(% style="width:105px" %)Клиент|(% style="width:1370px" %)1) От клиентов поступают заявки на некоторую обработку.
14 На основе заявки создается запись (State) в потокобезопасной коллекции ([[ConcurrentDictionary>>doc:Разработка.NET.C#.Коллекции.System\. Collections\. Concurrent.ConcurrentDictionary.WebHome]]). (В записи также содержится CancalidationToken, позволяющий прервать await клиента)
15 Запрос встает на ожидание обработки с await. (LongPooling не разрываем соединение с клиентом)
16 |(% style="width:105px" %)Воркер|(% style="width:1370px" %)2) Через некоторые промежутки времени запускается некоторый [[Концепция Worker (Worker and WorkerParallelWrapper)>>doc:Архитектура и модели.Модели.Модель Worker (Worker and WorkerParallelWrapper).WebHome]], который берет группу заявок из потокобезопасной коллекции и начинает их обрабатывать.
17 По окончанию обработки Worker записывает результаты в соответствущие State и вызывает CancalidationToken, чтобы разблокировать ожидающего клиента.
18 |(% style="width:105px" %)Клиент|(% style="width:1370px" %)3) Клиент выходит с await и забирает свой State, в котором содержится некоторый результат обработки.
19 (
20 Статус успешно и результат
21 либо статус ошибка и сообщенеи об ошибке
22 )
23
24 ----
25
26 {{code language="C#"}}
27 public class BatchProcessingContainer<TRequest, TResponse>
28 {
29 private readonly ConcurrentDictionary<Guid, (TRequest Request, TResponse? Response, CancelationTokenSource WaitToken) _Buffer
30
31 public async Task<TResponse?> RequestAsync(
32 TRequest request,
33 CancelationToken token = default
34 );
35
36
37 public async ValueTask<(Guid RequestId, TRequest Request)[]> GetForProcessingAsync(
38 int count? = null
39 );
40
41 public async ValueTask SetResultAsync(
42 Guid requestId,
43 TResponse response
44 )
45 }
46 {{/code}}