launch в main лишний, не убрал. Пример взят с оф г...
# russiana
Alexjok
07/30/2018, 4:47 AMlaunch в main лишний, не убрал. Пример взят с оф гайда на гитхаб, идея простая, скажем в launchprocessor из produce продюсируем xml в launch там их разбираем и записываем в БД, на каждый produce и launch своя корутина. Описанный пример - моя фантазия, но пока не клеится с реализацией
g
gildor
07/30/2018, 5:00 AMя не вижу изменений, все равно launch в consumeEach и в launchProcessor, и главная пробелма с чтением двумя вложеными корутинами одного и того же канала все так же осталась
gildor
07/30/2018, 5:03 AMЕсли речь о “Fan-out” примере, то там все как раз верно, там не читаются данные из канала дважды, только процессорами. Там есть несколько процессоров запущеных, они все разом читают один и тот же канал и они получают данные и потом могут с ними что-то делать
В примере выше все не так, там сломано чтение из канала и данные сохраняются игнорируя тред сейф в мутабельную переменную.
Но да, пример про Fan-out может использоваться для юзкейса с процессингом
gildor
07/30/2018, 5:08 AMНа самом деле не понятно зачем в вашм примере нужен producer.consumeEach, так как он читает данные из канала и не является процессором
a
Alexjok
07/30/2018, 5:13 AMЯ правильно понимаю, что если заменить consumeEach на repeat с нужным количеством повторений, то все будет работать корректно?
g
gildor
07/30/2018, 5:17 AMчастично да.
Там есть еще проблемы
a
Alexjok
07/30/2018, 5:17 AMЧастично меня не устраивает))
g
gildor
07/30/2018, 5:18 AMНапример так как хочется запустить все процессоры паралельно, то нельзя будет просто на следующей строчке вызвать
println("Количество ${ok.size}")gildor
07/30/2018, 5:19 AMну и глобально, нужно как то забьлокировать завершение программы, что бы не усложнять пример с ожиданием всех job можно просто добавить delay
gildor
07/30/2018, 5:19 AMну или все же сложить все Job в лист и дождаться завершения каждой, что бы вывести результат
gildor
07/30/2018, 5:20 AMтак же остается проблема с тем что переменная ok не тред сейф и вообще мутабельный стейт
gildor
07/30/2018, 5:22 AMеще нужно убрать не только
producer.consumeEachlaunch(CommonPool)gildor
07/30/2018, 5:23 AMвы же уже запускаете новый процессор внутри launchProcessor, еще раз оборачивать его в корутину не нужно
a
Alexjok
07/30/2018, 5:27 AMНаверное с листом job'ов самое простое решение, delay не понятно как будет работать. В моем случае, какое-то время нужно на produce xml и потом какое-то на парсинг в launch, как правильно поставить delay в таком случае я не знаю. Нужно, чтобы job куда-то были подвязаны, подумать надо
Alexjok
07/30/2018, 5:27 AMИ ещё, как все таки consumeEach работает, я вот думал, что он весь поток продюсирует
g
gildor
07/30/2018, 5:27 AMпро delay я просто для примера
gildor
07/30/2018, 5:27 AMconsumeEach не продюсирует, он читает значения из канала
gildor
07/30/2018, 5:28 AMт.е. сейчас спрева читаете значение из канала, потом этот же канал отдаете процессору, где он тоже уже следующие значение читает
a
Alexjok
07/30/2018, 5:28 AMТеперь понятно почему примерно половину пишет, спасибо
g
gildor
07/30/2018, 5:30 AMЧто-то вроде этого должно у вас получится, я поправил замеченые проблемы и замнил лист результатов просто атомик каунтером, в реальности вы будете писать в файл или базу и там не должно быть проблем
https://gist.github.com/gildor/bcc5cad08f6b7226f6bba98778bee99e
a
Alexjok
07/30/2018, 5:32 AMАндрей, спасибо большое за ваши ответы) Пойду ковыряться :)
g
gildor
07/30/2018, 5:33 AMgildor
07/30/2018, 5:34 AMтам только старая версия kotlinx.coroutines, поэтому
.joinAll().forEach { it.join() }a
Alexjok
07/30/2018, 5:37 AMпо улице с телефоном иду) но да работает как я себе представлял, спасибо вам
g
gildor
07/30/2018, 5:54 AMКак вариант проэмулировать хранилище можно через актор, который будет писать данные в одном потоке:
Copy code
val dbData = mutableListOf<Int>()
val actorDb = actor<Int> {
for (value in channel) {
// Эмулируем задержку записи
delay(20)
dbData += value
}
} Copy code
actorDb.send(msg)a
Alexjok
07/30/2018, 6:02 AMЯ до actor ещё не дочитал, меня на channels понесло:)
g
gildor
07/30/2018, 6:03 AMну актор и есть обертка над каналом, так что правильно что про каналы нужно сперва почитать
gildor
07/30/2018, 6:03 AMактор в данном примере как таковой не нужен, через него тоже можно сделать, но проще через то как это сделано с голым каналом, просто через актор удобно проэмулировать работу с базой или любым другим асинхронным хранилищем
gildor
07/30/2018, 6:04 AMОбновил пример с использованием актора для сохранения данных вместо просто каунтера
a
Alexjok
07/30/2018, 6:07 AMСегодня прикручу реальные данные и парсинг с последующим сохранением в БД, думаю будет круто работать
g
gildor
07/30/2018, 6:09 AMно имейте в виду, так как “база” в виде актора одно-поточная, то есть запись всех значений не может быть быстрее чем сумарное время записи всех элементов в одном потоке, с этим тоже можно поиграться, эмулируя реальную базу (меняя capacity у актора, это кеш входящих знаений или меняя задержку на запись)
gildor
07/30/2018, 6:11 AMТолько имейте в виду, что если сохранение в базу и парсинг блокирующие операции, они будут блокировать CommonPool, что не хорошо, лучше или что бы они были асинхронные или выполнялись на своих диспатчерах
a
Alexjok
07/30/2018, 6:31 AMв моем кейсе, мы из продюсера отправляем в парсер кусочек xml, затем распаршенный кусок отправляем в, что-то типа валидатора, там так же работаем над данными, какие-то значения меняем и затем делаем сейв в БД. На самом деле парсер тоже своего рода produce, который данные дальше по цепочке передаёт дёргая оттуда, что нужно, валидатор финальная точка с сохранением в базе. Я думаю на каждый кусок запускать по job, то есть сколько у нас кусков xml будет продюсироваться, столько процессов и создавать.
2 Views