Исходный код вики Пакетная обработка заявок с асинхронностью (Async)

Редактировал(а) Alexandr Fokin 2023/01/23 23:00

version	line-number	content
5.1	1	[[Асинхронность. Async Await>>doc:Разработка.NET.C#.Многопоточность и асинхронность.Асинхронность\. Async Await.WebHome]]
1.1	2
6.2	3	[[Модели>>doc:Архитектура и модели.Модели.WebHome]]
3.4	4
	5	----
	6
3.1	7	Используемые инструменты
	8	1) Пакетная обработка
	9	2) LongPooling
	10
1.1	11
3.6	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 не разрываем соединение с клиентом)
6.1	16	\|(% style="width:105px" %)Воркер\|(% style="width:1370px" %)2) Через некоторые промежутки времени запускается некоторый [[Концепция Worker (Worker and WorkerParallelWrapper)>>doc:Архитектура и модели.Модели.Модель Worker (Worker and WorkerParallelWrapper).WebHome]], который берет группу заявок из потокобезопасной коллекции и начинает их обрабатывать.
3.6	17	По окончанию обработки Worker записывает результаты в соответствущие State и вызывает CancalidationToken, чтобы разблокировать ожидающего клиента.
	18	\|(% style="width:105px" %)Клиент\|(% style="width:1370px" %)3) Клиент выходит с await и забирает свой State, в котором содержится некоторый результат обработки.
2.1	19	(
3.6	20	Статус успешно и результат
2.1	21	либо статус ошибка и сообщенеи об ошибке
3.6	22	)
3.5	23
3.6	24	----
	25
3.7	26	{{code language="C#"}}
3.8	27	public class BatchProcessingContainer<TRequest, TResponse>
3.7	28	{
	29	private readonly ConcurrentDictionary<Guid, (TRequest Request, TResponse? Response, CancelationTokenSource WaitToken) _Buffer
	30
3.9	31	public async Task<TResponse?> RequestAsync(
3.7	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}}