#arrays #spring-webflux #project-reactor
#массивы #spring-webflux #проект-реактор
Вопрос:
У меня есть следующий сценарий для загрузчика данных в приложении Webflux с использованием реактивного драйвера MongoDB и Spring:
- создать X объектов типа B
- создайте Y объектов типа A: объект A содержит поле типа array и ссылку на объект типа B. Ссылка на B выбирается случайным образом с первого шага
- добавьте N записей в массив ранее созданного объекта
Проблема, с которой я сталкиваюсь, по-видимому, заключается в параллельном выполнении Mono / Flux, чего, насколько я понимаю, не должно происходить. Согласно документации все всегда выполняется последовательно, если не указано иное.
Может кто-нибудь, пожалуйста, дать мне подсказку, что я делаю неправильно?
Вот пример фрагмента кода. Объект A — это туалет. Объект B является пользователем. Поле массива — это поле комментариев:
Flux.range(0, 10)
// create 10 objects of type user
.flatMap {
LOG.debug("Creating user $it")
userRepository.save(
User(
id = ObjectId(),
name = userNames.random(),
email = "${userNames.random()}@mail.com"
)
)
}
.collectList()
// create 2 objects of type toilet
.flatMapMany { userList ->
Flux.range(0, 2).zipWith(Flux.range(0, 2).map { userList })
}
.flatMap {
LOG.debug("Creating toilet ${it.t1}")
val userList = it.t2
toiletRepository.save(
Toilet(
id = ObjectId(),
title = userList.random().name
)
)
}
// add 5 entries to array of toilet
.flatMap { toilet ->
Flux.range(0, 5).zipWith(Flux.range(0, 5).map { toilet })
}
.flatMap { tuple ->
val toilet = tuple.t2
LOG.debug("Creating comment ${tuple.t1} for toilet $toilet")
// get current values from toilet
toiletRepository.findById(toilet.id).map {
// and push a new element to the comments array
LOG.debug("Comment size ${it.commentRefs.size}")
toiletRepository.save(it.apply { commentRefs.add(ObjectId()) })
}
}
.subscribe {
GlobalScope.launch {
exitProcess(SpringApplication.exit(context))
}
}
Выполнение этого кода приводит к следующему журналу:
2020-11-15 19:42:54.197 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 0
2020-11-15 19:42:54.293 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 1
2020-11-15 19:42:54.295 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 2
2020-11-15 19:42:54.296 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 3
2020-11-15 19:42:54.300 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 4
2020-11-15 19:42:54.301 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 5
2020-11-15 19:42:54.304 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 6
2020-11-15 19:42:54.310 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 7
2020-11-15 19:42:54.316 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 8
2020-11-15 19:42:54.318 DEBUG 13524 --- [ main] c.g.q.t.DataLoaderRunner : Creating user 9
2020-11-15 19:42:54.348 DEBUG 13524 --- [tLoopGroup-3-10] c.g.q.t.DataLoaderRunner : Creating toilet 0
2020-11-15 19:42:54.380 DEBUG 13524 --- [tLoopGroup-3-10] c.g.q.t.DataLoaderRunner : Creating toilet 1
2020-11-15 19:42:54.386 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 0 for toilet Toilet(id=5fb176aef24f4c248fbb051c, title=wholesale, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.405 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 1 for toilet Toilet(id=5fb176aef24f4c248fbb051c, title=wholesale, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.406 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 2 for toilet Toilet(id=5fb176aef24f4c248fbb051c, title=wholesale, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.407 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 3 for toilet Toilet(id=5fb176aef24f4c248fbb051c, title=wholesale, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.409 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 4 for toilet Toilet(id=5fb176aef24f4c248fbb051c, title=wholesale, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.410 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 0 for toilet Toilet(id=5fb176aef24f4c248fbb051d, title=imaginary, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.412 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 1 for toilet Toilet(id=5fb176aef24f4c248fbb051d, title=imaginary, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.413 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 2 for toilet Toilet(id=5fb176aef24f4c248fbb051d, title=imaginary, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.414 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 3 for toilet Toilet(id=5fb176aef24f4c248fbb051d, title=imaginary, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.415 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Creating comment 4 for toilet Toilet(id=5fb176aef24f4c248fbb051d, title=imaginary, location=Point [x=0.000000, y=0.000000], previewID=null, averageRating=0.0, ratingRefs=[], disabled=false, toiletCrewApproved=false, description=, commentRefs=[], imageRefs=[])
2020-11-15 19:42:54.425 DEBUG 13524 --- [tLoopGroup-3-10] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-8] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-6] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-2] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-3] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-9] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.425 DEBUG 13524 --- [ntLoopGroup-3-7] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.429 DEBUG 13524 --- [ntLoopGroup-3-2] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.429 DEBUG 13524 --- [ntLoopGroup-3-9] c.g.q.t.DataLoaderRunner : Comment size 0
2020-11-15 19:42:54.464 DEBUG 13524 --- [tLoopGroup-3-10] c.g.q.t.DataLoaderRunner : Comment size 0
Теперь у меня есть три вопроса:
- Почему поток переключается с main на LoopGroup? Если он выполняется в последовательности, он вообще не должен использовать многопоточность?
- Почему
Comment sizeсообщения журнала группируются вместе в конце? - Как правильно помещать элементы в массив, используя реализацию репозитория reactive mongo?
Любые подсказки приветствуются. Я предполагаю, что вложенное выполнение findById и save неверно, но как бы вы написали это по-другому? Поскольку save требуется объект, мне нужно передать последнюю версию объекта, которая содержит один дополнительный элемент в массиве. Я пытаюсь достичь этого, получив последнюю версию findById и непосредственно изменив ее с помощью ‘map -> save’.
Спасибо вам всем!
Ответ №1:
Я не уверен, что это лучший способ сделать это, но я смог добиться того, чего я хочу, разделив операции в функциях, чтобы они были сгруппированы более логично.
Вот фрагменты для следующих операций:
- создание пользователей
- создавать комментарии
- создание рейтингов
private fun createUsers() = Flux.range(0, userNames.size 1)
.flatMap {
if (it < userNames.size) {
LOG.debug("Creating user $it")
userRepository.save(
User(
id = ObjectId(),
name = userNames[it],
email = "${userNames[it]}@mail.com"
)
)
} else {
LOG.debug("Creating dev-user")
userRepository.save(
User(
id = ObjectId("000000000000012343456789"),
name = "devuser",
email = "devuser@mail.com"
)
)
}
}
.collectList()
private fun createComments(users: List<User>) = Flux.range(0, numComments)
.flatMap {
LOG.debug("Creating comment $it")
commentRepository.save(
Comment(
id = ObjectId(),
text = commentTexts.random(),
userRef = users.random().id
)
)
}
.collectList()
private fun createRatings(users: List<User>) = Flux.range(0, numRatings)
.flatMap {
LOG.debug("Creating rating $it")
ratingRepository.save(
Rating(
id = ObjectId(),
userRef = users.random().id,
value = Random.nextInt(0, 5)
)
)
}
.collectList()
И, наконец, создание туалетов с результатом сверху:
private fun createToilets(comments: List<Comment>, ratings: List<Rating>) = Flux.range(0, numToilets)
.flatMap {
val toilet = Toilet(
id = ObjectId(),
title = titles.random(),
location = GeoJsonPoint(Random.nextDouble(10.0, 20.0), Random.nextDouble(45.0, 55.0)),
description = descriptions.random()
)
// add comments
val commentsToAdd = Random.nextInt(0, comments.size)
for (i in 0 until commentsToAdd) {
toilet.commentRefs.add(comments[i].id)
}
// add average rating and rating references
val ratingsToAdd = Random.nextInt(0, ratings.size)
for (i in 0 until ratingsToAdd) {
toilet.ratingRefs.add(ratings[i].id)
toilet.averageRating = ratings[i].value
}
if (toilet.ratingRefs.isNotEmpty()) {
toilet.averageRating /= toilet.ratingRefs.size
}
LOG.debug("Creating toilet $it with $commentsToAdd comments and $ratingsToAdd ratings")
toiletRepository.save(toilet)
}
// upload preview image
.flatMap { toilet ->
val imageName = "toilet${Random.nextInt(1, 10)}.jpg"
imageService.store(
Callable {
DataLoaderRunner::class.java.getResourceAsStream("/sample-images/$imageName")
},
"${toilet.title}-preview"
).zipWith(Mono.just(toilet))
}
// set preview image
.flatMap {
val imageId = it.t1
val toilet = it.t2
toiletRepository.save(toilet.copy(previewID = imageId))
}
.collectList()
Это последняя реактивная цепочка операций:
createUsers()
.flatMap { users ->
createComments(users).map { comments ->
Tuples.of(users, comments)
}
}
.flatMap {
val users = it.t1
val comments = it.t2
createRatings(users).map { ratings ->
Tuples.of(comments, ratings)
}
}
.flatMap {
val comments = it.t1
val ratings = it.t2
createToilets(comments, ratings)
}
// close application when all toilets are processed
.subscribe {
GlobalScope.launch {
exitProcess(SpringApplication.exit(context))
}
}
Я не уверен, что это лучший способ сделать это, но он работает. Подход, описанный во вступительном сообщении, заключается в использовании вложенных операций map / flatmap, которых в любом случае следует избегать, и, возможно, они являются причиной того, что это не сработало.