Всем привет, у меня есть обработчик, для которого нужно выполнить следующую функциональность.
1) Получить список ID-документов через http-запрос
2) Для каждого ID документа сделать http-запрос и распарсить его.
3) Получить распарсенные данные из всех документов.
Я реализую многопоточную обработку посредством использования System.Threading.Channels.
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
// Главная точка входа
public async Task ProcessDocsAsync()
{
//канал для получения и отправки ID-документов
var docIDChannel = Channel.CreateUnbounded<Doc>(new UnboundedChannelOptions() { SingleReader = false,SingleWriter = true });
//канал, куда складываем значения из распарсенных JSON-документов
var parseChannel = Channel.CreateUnbounded<(Doc docID, string uid)>();
var taskList = new List<Task>();
//читаем номера документов посредством http-запроса
taskList
.Add(Task.Run(async () => await GetDocsNumbersAsync(docIDChannel.Writer, _apiProcessorSettings.GetDocsRoute()))
.ContinueWith(t =>
{
docIDChannel.Writer.Complete();
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("DocReading Complete");
Console.ForegroundColor = ConsoleColor.White;
}));
//создаем 5 задач, которые будут распарсивать документы
var processList = Enumerable
.Range(1, 5)
.Select(_ => Task.Run(async () => await ParseDocByIsAsync(docIDChannel.Reader, parseChannel.Writer)));
//информируем канал о том, что больше добавления элементов не произойдет
var procTask = Task.WhenAll(processList).ContinueWith(t =>
{
parseChannel.Writer.Complete();
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("Parsing Complete");
Console.ForegroundColor = ConsoleColor.White;
});
taskList.Add(procTask);
// читаем распарсенные документы
taskList
.Add(Task.Run(async () => await ConsumeOutputProductsInConsoleAsync(parseChannel.Reader)));
await Task.WhenAll(taskList);
Console.WriteLine("Done");
}
//Получение документов
private async Task GetDocsNumbersAsync(ChannelWriter<Doc> channelWriter, string route)
{
var сlient = _httpClientFactory.CreateClient("сlient");
using var docStream = await сlient.GetStreamAsync(route);
using JsonDocument jDocument = await JsonDocument.ParseAsync(docStream);
var root = jDocument.RootElement.GetProperty("docs").EnumerateArray();
foreach (var item in root)
{
await channelWriter.WriteAsync(new Doc(
item.GetProperty("number").GetString()));
}
}
//парсинг json-документов
private async Task ParseDocByIsAsync(ChannelReader<Doc> docChannel,
ChannelWriter<(Doc doc, string uid)> uitChannel)
{
while (await docChannel.WaitToReadAsync())
{
while (docChannel.TryRead(out var doc))
{
try
{
Console.WriteLine($"start processing doc:{doc.DocID}, thread:{Thread.CurrentThread.ManagedThreadId}");
int accumulateCount = 0;
var сlient = _httpClientFactory.CreateClient("сlient");
var response = await сlient.GetAsync($"route/{doc.DocID}/", HttpCompletionOption.ResponseHeadersRead);
using Stream streamToReadFrom = await response.Content.ReadAsStreamAsync();
using JsonDocument jDocument = await JsonDocument.ParseAsync(streamToReadFrom);
var root = jDocument.RootElement;
//количество
var allCount = root.GetProperty("total").GetInt32(); /// вот здесь ошибка при многопоточном использовании
//парсим документ и в итоге отправляем распарсенные значение в канал
///...
await uitChannel.WriteAsync((doc, uit));
}
catch (Exception ex)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("Error: " + ex.Message);
Console.ForegroundColor = ConsoleColor.White;
}
}
}
}
// чтение распарсенных документов
private async Task ConsumeOutputProductsInConsoleAsync(ChannelReader<(Doc doc, string uid)> uitChannel)
{
long cnt = 0;
while (await uitChannel.WaitToReadAsync())
{
if (uitChannel.TryRead(out var doc))
{
cnt++;
}
}
Console.WriteLine($"Count is:{cnt}");
}
Для вышеуказаного сниппета я получаю нежелательные side-эффекты, которые отсутствуют если я запускаю парсинг в режиме одной задачи
1.
2.
3.
var processList = Enumerable
.Range(1, 1)
.Select(_ => Task.Run(async () => await ParseDocByIsAsync(docIDChannel.Reader, parseChannel.Writer)));
В методе ParseDocByIsAsync я делаю http-запрос, в результате которого возвращается JSON- документ, у которого в обязательном порядке есть свойство "total"- проверял много раз через Postman.
В результате выполнения кода я попадаю в секцию "Exception" с текстом ошибки об отсутствии свойства/токена "total" у Json-документа.
Я проверяю url, по которому запрашивается Json- он приходит корректный. Проверяю значение переменной "root" при возникновении ошибки- действительно именно это свойство отсутствует. Как будто один поток прочитал свойство, закорраптил его, положил в пул для дальнейшего выполнения, другой поток подхватил таску и у же не может его нормально прочитать.
Ошибка носит вероятностный характер и её вероятность возрастает с увеличением количества Task для обработки парсинга. При 5 тасках на парсинг вероятность ошибки около 90%, хотя бывало и успешное выполнение.
Также заметил, что один документ может обрабатываться несколькими потоками- возможно это просто перекладывание таски из одного в другой, но использование Channels обещает потокобезопасное получение элементов из канала и то что элемент не может быть получен срузу несколькими потоками. Хоть бери и на BlockCollection откатывайся. Подскажите, кто с такой ситуацией сталкивался. Спасибо
