{"id":341197,"date":"2022-11-14T03:00:12","date_gmt":"2022-11-14T03:00:12","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=341197"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=341197","title":{"rendered":"\u041f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u0438\u0437\u043c, \u0430\u0441\u0438\u043d\u0445\u0440\u043e\u043d\u043d\u043e\u0441\u0442\u044c, \u043c\u043d\u043e\u0433\u043e\u043f\u043e\u0442\u043e\u0447\u043d\u043e\u0441\u0442\u044c \u2013 Reactor \u043f\u043e\u0447\u0442\u0438 \u0432\u0441\u0451 \u0441\u0434\u0435\u043b\u0430\u0435\u0442 \u0437\u0430 \u0412\u0430\u0441<\/span>"},"content":{"rendered":"
<\/div>\n
\n
\n
\n
\n

\u0426\u0435\u043b\u044c\/\u0432\u0432\u0435\u0434\u0435\u043d\u0438\u0435<\/h2>\n

\u0420\u0435\u0430\u043a\u0442\u0438\u0432\u043d\u044b\u0435 \u043f\u0430\u0442\u0442\u0435\u0440\u043d\u044b \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u0441\u0442\u0430\u043d\u043e\u0432\u044f\u0442\u0441\u044f \u0432\u0441\u0451 \u0431\u043e\u043b\u0435\u0435 \u0432\u043e\u0441\u0442\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u044b \u043f\u0440\u0438 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0432\u044b\u0441\u043e\u043a\u043e\u043d\u0430\u0433\u0440\u0443\u0436\u0435\u043d\u043d\u044b\u0445 \u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432. \u0420\u0435\u0430\u043a\u0442\u0438\u0432\u043d\u044b\u0435 \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u0438 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u044e\u0442 \u0438\u043d\u0441\u0442\u0440\u0443\u043c\u0435\u043d\u0442\u044b, \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u044e\u0449\u0438\u0435 \u0441 \u043c\u0438\u043d\u0438\u043c\u0430\u043b\u044c\u043d\u044b\u043c\u0438 \u0437\u0430\u0442\u0440\u0430\u0442\u0430\u043c\u0438 \u043d\u0430 \u043a\u043e\u0434\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043c\u0435\u0445\u0430\u043d\u0438\u0437\u043c\u044b \u0430\u0441\u0438\u043d\u0445\u0440\u043e\u043d\u043d\u043e\u0441\u0442\u0438 \u0438 \u043c\u043d\u043e\u0433\u043e\u043f\u043e\u0442\u043e\u0447\u043d\u043e\u0441\u0442\u0438.<\/p>\n

\u0412 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u0430, \u043f\u0440\u0435\u0434\u043b\u0430\u0433\u0430\u044e \u0440\u0430\u0441\u0441\u043c\u043e\u0442\u0440\u0435\u0442\u044c \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044e \u0441\u0435\u0440\u0432\u0438\u0441 \u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 ElasticSearch. \u0414\u0430\u043d\u043d\u044b\u0435 \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0432 MongoDB, \u043a\u043b\u044e\u0447\u0435\u0432\u044b\u0435 \u0430\u0442\u0440\u0438\u0431\u0443\u0442\u044b \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0441\u0438\u043d\u0445\u0440\u043e\u043d\u0438\u0437\u0438\u0440\u0443\u044e\u0442\u0441\u044f \u0441 ElasticSearch (\u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u043e \u043f\u043e\u0445\u043e\u0436\u0435 \u043d\u0430 Logstash). \u0412 \u043f\u0440\u043e\u0435\u043a\u0442\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u0442\u0435\u043a: Java\/Spring Boot\/Reactor\/WebFlux\/WebClient\/RabbitMQ\/MongoDB. \u041d\u0430 \u0432\u044b\u0431\u043e\u0440 RabbitMQ \u0438 MongoDB \u043f\u043e\u0432\u043b\u0438\u044f\u043b\u043e, \u0432 \u0442\u043e\u043c \u0447\u0438\u0441\u043b\u0435, \u043d\u0430\u043b\u0438\u0447\u0438\u0435 \u0440\u0435\u0430\u043a\u0442\u0438\u0432\u043d\u044b\u0445 \u0434\u0440\u0430\u0439\u0432\u0435\u0440\u043e\u0432.<\/p>\n

\u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0437\u0430\u0434\u0430\u0447\u0438<\/h3>\n
    \n
  1. \n

    \u0421\u0435\u0440\u0432\u0438\u0441 \u0434\u043e\u043b\u0436\u0435\u043d \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u0442\u044c \u043f\u043e\u0442\u043e\u043a \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u043e\u0447\u0435\u0440\u0435\u0434\u0438, \u0432\u044b\u0431\u0438\u0440\u0430\u0442\u044c \u0441\u0432\u044f\u0437\u0430\u043d\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u0431\u0430\u0437\u044b \u0438 \u043f\u0435\u0440\u0435\u0434\u0430\u0432\u0430\u0442\u044c \u0438\u0445 ElasticSearch. \u0424\u043e\u0440\u043c\u0430\u0442 \u0434\u0430\u043d\u043d\u044b\u0445 \u043e\u0447\u0435\u0440\u0435\u0434\u0438: \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0435 (index\/delete); id \u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430; \u0438\u043c\u044f \u0438\u043d\u0434\u0435\u043a\u0441\u0430; \u0442\u0438\u043f \u0438\u043d\u0434\u0435\u043a\u0441\u0430 (\u043e\u043f\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u043e).<\/p>\n<\/li>\n

  2. \n

    \u0427\u0435\u0440\u0435\u0437 web-\u0438\u043d\u0442\u0435\u0440\u0444\u0435\u0439\u0441 \u0434\u043e\u043b\u0436\u0435\u043d \u0431\u044b\u0442\u044c \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d \u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b \u0434\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u044f, \u0443\u0434\u0430\u043b\u0435\u043d\u0438\u044f \u0438 \u043f\u0435\u0440\u0435\u0441\u0442\u0440\u043e\u0435\u043d\u0438\u044f \u0438\u043d\u0434\u0435\u043a\u0441\u0430.<\/p>\n<\/li>\n

  3. \n

    \u0414\u043e\u043b\u0436\u043d\u0430 \u0431\u044b\u0442\u044c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u0444\u043e\u0440\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u0430\u0433\u0440\u0435\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 \u043f\u043e\u043b\u0435\u0439, \u0441\u043e\u0434\u0435\u0440\u0436\u0430\u0449\u0438\u0445 \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u0438\u0445 \u0438\u0441\u0445\u043e\u0434\u043d\u044b\u0445 \u043f\u043e\u043b\u0435\u0439 \u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u043e\u0432, \u0438 \u0434\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0438\u043d\u0434\u0435\u043a\u0441 \u0438\u0437 \u0441\u0432\u044f\u0437\u0430\u043d\u043d\u044b\u0445 \u043a\u043e\u043b\u043b\u0435\u043a\u0446\u0438\u0439.<\/p>\n<\/li>\n

  4. \n

    \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0438\u043d\u0434\u0435\u043a\u0441\u0438\u0440\u0443\u0435\u043c\u044b\u0445 \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u0442\u044c \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 JSON.<\/p>\n<\/li>\n<\/ol>\n

    DFD-\u0434\u0438\u0430\u0433\u0440\u0430\u043c\u043c\u0430 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430 \u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u0438<\/h2>\n

    \u0421\u0445\u0435\u043c\u0430 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430 \u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u0438 \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432, \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0445 \u0438\u0437 \u043e\u0447\u0435\u0440\u0435\u0434\u0438, \u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c:<\/p>\n

    <\/figcaption><\/figure>\n

    \u0410\u043b\u0433\u043e\u0440\u0438\u0442\u043c \u043f\u0435\u0440\u0435\u0441\u0442\u0440\u043e\u0435\u043d\u0438\u044f \u0438\u043d\u0434\u0435\u043a\u0441\u0430 \u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442 \u043f\u0440\u0430\u043a\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0442\u0430\u043a\u0436\u0435, \u0437\u0430 \u0438\u0441\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0435\u043c \u0442\u043e\u0433\u043e, \u0447\u0442\u043e \u0432 \u043d\u0435\u043c \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u0443\u0435\u0442 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0430 \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432, \u043e\u0442\u043b\u043e\u0436\u0435\u043d\u043d\u044b\u0445 \u0438\u0437-\u0437\u0430 \u043e\u0448\u0438\u0431\u043e\u043a.<\/p>\n

    \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b\u0430<\/h2>\n

    \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b\u0430 \u043a\u043e\u0441\u043d\u0435\u0442\u0441\u044f \u0442\u043e\u043b\u044c\u043a\u043e \u0440\u0430\u0431\u043e\u0442\u044b \u0440\u0435\u0430\u043a\u0442\u0438\u0432\u043d\u043e\u0439 \u0447\u0430\u0441\u0442\u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u0430. \u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u043e\u043d\u043d\u044b\u0435 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438, \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0430 \u0444\u043e\u0440\u043c\u0430\u0442\u0430 \u043e\u043f\u0438\u0441\u0430\u043d\u0438\u044f \u0438\u043d\u0434\u0435\u043a\u0441\u0438\u0440\u0443\u0435\u043c\u044b\u0445 \u0434\u0430\u043d\u043d\u044b\u0445, \u0444\u043e\u0440\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043b\u044f \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 \u043a ElasticSearch \u0432\u044b\u043d\u0435\u0441\u0435\u043d\u044b \u0437\u0430 \u0440\u0430\u043c\u043a\u0438 \u0434\u0430\u043d\u043d\u043e\u0439 \u0441\u0442\u0430\u0442\u044c\u0438, \u043d\u043e \u0432\u044b \u043c\u043e\u0436\u0435\u0442\u0435 \u043f\u043e\u0441\u043c\u043e\u0442\u0440\u0435\u0442\u044c \u043a\u043e\u0434 \u043d\u0430 GitHub, \u043f\u043e \u0441\u0441\u044b\u043b\u043a\u0435<\/a>.<\/p>\n

    \u0422\u0435\u043f\u0435\u0440\u044c \u043f\u043e\u043f\u0440\u043e\u0431\u0443\u0435\u043c \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u044d\u0442\u0443 \u0441\u0445\u0435\u043c\u0443 \u0441\u043a\u0432\u043e\u0437\u043d\u044b\u043c \u043f\u043e\u0442\u043e\u043a\u043e\u043c Reactor, \u043d\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043f\u043e\u0434\u043f\u0438\u0441\u043a\u0438 \u043d\u0430 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u044d\u043b\u0435\u043c\u0435\u043d\u0442\u044b, \u0432 \u0442\u043e\u043c \u0447\u0438\u0441\u043b\u0435 \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0443 \u0447\u0435\u0440\u0435\u0437 WebClient HTTP-\u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 \u0438 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0443 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043d\u044b\u0445 \u043e\u0442\u0432\u0435\u0442\u043e\u0432. \u041e\u0442\u0434\u0430\u0434\u0438\u043c, \u043f\u043e\u0447\u0442\u0438 \u043f\u043e\u043b\u043d\u043e\u0441\u0442\u044c\u044e, \u0441\u0438\u043d\u0445\u0440\u043e\u043d\u0438\u0437\u0430\u0446\u0438\u044e \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u044f Reactor. <\/p>\n

    \u041a\u043e\u0434, \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u044e\u0449\u0438\u0439 \u043f\u0440\u043e\u0446\u0435\u0441\u0441 \u043f\u0435\u0440\u0435\u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u0438 \u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c: <\/p>\n

    Task task = new Task(mongoElasticIndex); ParallelFlux dataEventsFlux = reactorRepositoryMongoDB         .findAll(mongoElasticIndex.getCollection(), mongoElasticIndex.getProjection())         .parallel(appConfig.getIndexParallelism())         .runOn(Schedulers.boundedElastic()); Flux<Tuple2<String,Document>> processingData = processingData(dataEventsFlux, (p) -> \"index\",         (p) -> (Document)p,         (p) -> mongoElasticIndex,         Flux.just(),         task);  task.setDispose(subscribe(processedData, task));<\/code><\/pre>\n
    \u041f\u043e\u043b\u0443\u0447\u0430\u0435\u043c \u043f\u043e\u0442\u043e\u043a \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u043a\u043e\u043b\u043b\u0435\u043a\u0446\u0438\u0438, \u043d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0435\u043c \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u0438\u0437\u043c,  \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u043c \u043e\u0431\u044a\u0435\u043a\u0442 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u043f\u043e\u0442\u043e\u043a\u0430 \u0438 \u043f\u043e\u0434\u043f\u0438\u0441\u044b\u0432\u0430\u0435\u043c\u0441\u044f \u043d\u0430 \u043f\u043e\u0442\u043e\u043a. \u0417\u0434\u0435\u0441\u044c \u043a\u043b\u0430\u0441\u0441 Task \u2013 \u0432\u043d\u0443\u0442\u0440\u0435\u043d\u043d\u0438\u0439 \u043a\u043b\u0430\u0441\u0441, \u043d\u0430\u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e: \u0441\u043e\u0431\u0438\u0440\u0430\u0442\u044c \u0441\u0442\u0430\u0442\u0438\u0441\u0442\u0438\u043a\u0443 \u0438 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044e \u043e \u0432\u044b\u043f\u043e\u043b\u043d\u044f\u0435\u043c\u044b\u0445 \u0437\u0430\u0434\u0430\u0447\u0430\u0445 \u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u0438.<\/p>\n
    \u041c\u0435\u0442\u043e\u0434 processingData \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u0442 \u043f\u043e\u0442\u043e\u043a \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 \u0438 \u043e\u0442\u0432\u0435\u0442\u043e\u0432, \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u0435\u043d\u043d\u044b\u0445 WebClient\u2019\u043e\u043c:<\/p>\n
    private <T> Flux<Tuple2<String,Document>>     processingData(ParallelFlux<T> events,             Function<T, String> getAction,                    Function<T, Document> getDocument,                    Function<T, MongoElasticIndex> getMongoElasticIndex,                    Flux<String> mergeFlux,                    Task task) {     return  events             \/\/ \u0414\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u043a \u0438\u0441\u0445\u043e\u0434\u043d\u043e\u043c\u0443 \u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0443 \u0438\u0437 \u043f\u0440\u0438\u0441\u043e\u0435\u0434\u0438\u043d\u044f\u0435\u043c\u044b\u0445 \u043a\u043e\u043b\u043b\u0435\u043a\u0446\u0438\u0439         .transform(joinData(getDocument, getMongoElasticIndex))             \/\/ \u0413\u0435\u043d\u0435\u0440\u0430\u0446\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043b\u044f \u043f\u0435\u0440\u0435\u0434\u0430\u0447\u0438 \u0432 ElasticSearch         .transform(document2ElasticJson(getAction, getDocument, getMongoElasticIndex))         .sequential()             \/\/ \u0410\u0433\u0440\u0435\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0434\u043b\u044f _bulk         .transform(grouping(task))             \/\/ \u0414\u043e\u0431\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u043f\u043e\u0442\u043e\u043a\u0430 \u0434\u0430\u043d\u043d\u044b\u0445, \u043d\u0430 \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043d\u0435 \u043f\u043e\u043b\u0443\u0447\u0435\u043d \u043e\u0442\u0432\u0435\u0442 \u043e\u0442 ElasticSearch         .mergeWith(mergeFlux)             \/\/ \u041e\u0442\u043f\u0440\u0430\u0432\u043a\u0430 \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 \u0432 ElasticSearch         .transform(postBulk(task))         .subscribeOn(Schedulers.single())         .doOnNext(testAliveResponses(task))         .doOnSubscribe(p-> p.request(appConfig.getMaxSizeBuffer() * 2))         .doOnComplete(() -> { logger.info(\"Start: {} End: {} read {} write {}\",                 formatDate(task.getStartDate()),                 formatDate(new Date()),                 task.getDocumentsRead(),                 task.getIndexesWrite(), getMaxProcessingRequest());             fileStorage.writeCollection2Files(waitingForResponse);             removeTask(task);         }); }<\/code><\/pre>\n
    \u041c\u0435\u0442\u043e\u0434\u043e\u043c transform Reactor \u0441\u043e\u0435\u0434\u0438\u043d\u044f\u0435\u043c \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0438 \u043f\u043e\u0442\u043e\u043a\u043e\u0432. \u0417\u0434\u0435\u0441\u044c \u0435\u0441\u0442\u044c \u043e\u0434\u043d\u043e \u0441\u0443\u0449\u0435\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0435 \u043e\u0433\u0440\u0430\u043d\u0438\u0447\u0435\u043d\u0438\u0435: \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0439 \u0438 \u0438\u0441\u0445\u043e\u0434\u044f\u0449\u0438\u0439 \u043f\u043e\u0442\u043e\u043a\u0438 \u0434\u043e\u043b\u0436\u043d\u044b \u0431\u044b\u0442\u044c \u043e\u0434\u043d\u043e\u0442\u0438\u043f\u043d\u044b\u043c\u0438 (Flux \u0438\u043b\u0438 ParallelFlux). \u041d\u0435\u043b\u044c\u0437\u044f, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e transform \u0432\u0441\u0442\u0440\u043e\u0438\u0442\u044c \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a \u0443 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u0432\u0445\u043e\u0434 Flux, \u0430 \u0432\u044b\u0445\u043e\u0434 ParallelFlux.<\/p>\n
    \u0412 \u043c\u0435\u0442\u043e\u0434 subscribe \u0441\u0435\u0440\u0432\u0438\u0441\u0430 \u0438\u043d\u043a\u0430\u043f\u0441\u0443\u043b\u0438\u0440\u043e\u0432\u0430\u043d\u0430 \u043f\u043e\u0434\u043f\u0438\u0441\u043a\u0430 \u043d\u0430 \u043f\u043e\u0442\u043e\u043a. \u041d\u0438\u0436\u0435 \u043f\u0440\u0438\u0432\u0435\u0434\u0435\u043d\u0430 \u0435\u0433\u043e \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f:<\/p>\n
    private Disposable subscribe(Flux<Tuple2<String,Document>> events, Task task) {     return  events         .subscribe(             p -> {                 if(isNull(task.getMongoElasticIndex())) { \/\/ \u0415\u0441\u043b\u0438 \u0437\u0430\u0434\u0430\u0447\u0430 \u043d\u0435 \u043f\u0435\u0440\u0435\u0438\u043d\u0434\u0435\u043a\u0441\u0430\u0446\u0438\u044f                     waitingForResponse.remove(p.getT1());                 }                 int count = Optional.ofNullable(p.getT2().get(\"items\", List.class))                         .map(List::size)                         .orElse(0);                 task.addIndexesWrite(count);             },             e -> {                 if(task != rabbitMQTask)removeTask(task);                 fileStorage.writeCollection2Files(waitingForResponse);                 logger.error(\"Error: {}\", e.getMessage());             }         ); }<\/code><\/pre>\n
    \u0414\u0430\u043b\u0435\u0435 \u043a\u043e\u0440\u043e\u0442\u043a\u043e \u043e\u0431 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0445 \u0444\u0443\u043d\u043a\u0446\u0438\u044f\u0445 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u043f\u043e\u0442\u043e\u043a\u0430.<\/p>\n
    \u0417\u0430\u0433\u0440\u0443\u0437\u043a\u0430 \u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u043e\u0432<\/h3>\n
    \u0418\u043c\u0435\u044e\u0442\u0441\u044f \u0434\u0432\u0430 \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0430 \u0437\u0430\u0433\u0440\u0443\u0437\u043a\u0438: <\/p>\n