rxjava 2: как утилизировать ресурс после того, как последующий процесс завершит обработку - PullRequest
Новая
       13

rxjava 2: как утилизировать ресурс после того, как последующий процесс завершит обработку

0 голосов
/ 30 июня 2019

У меня есть (Flowable) поток элементов, которые должны обрабатываться параллельно, используя один общий ресурс, и ресурс должен быть утилизирован впоследствии.Я пытался использовать оператор Single.using(), но он удаляет ресурс еще до обработки даже первого элемента в потоке.

Пример программы (на языке котлин): 

package my.test.rx_task_queue

import io.reactivex.Flowable
import io.reactivex.Single
import io.reactivex.schedulers.Schedulers
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicInteger

object TestCommonResource {
    private val logger = LoggerFactory.getLogger(TestCommonResource::class.java)
    @JvmStatic
    fun main(args: Array<String>) {
        val queue = Flowable.fromIterable(1..5)
        val resIdx = AtomicInteger(0)
        val resource = Single.using({
            val res = "resource-${resIdx.incrementAndGet()}"
            logger.info("Init resource $res")
            res
        }, { res ->
            Single.just(res)
        }, { res ->
            logger.info("Dispose resource $res")
        }, false)

        val result = resource.flatMap { res ->
            queue.flatMapSingle({ item ->
                Single.fromCallable {
                    logger.info("Process $item with $res")
                    "$item @ $res"
                }
                        .subscribeOn(Schedulers.io())
            }, false, 2)
                    .toList()
        }
                .blockingGet()
        logger.info("Result: $result")
    }
}

Пример вывода журнала: 

14:30:27.721 [main] INFO my.test.rx_task_queue.TestCommonResource - Init resource resource-1
14:30:27.744 [main] INFO my.test.rx_task_queue.TestCommonResource - Dispose resource resource-1
14:30:27.747 [RxCachedThreadScheduler-1] INFO my.test.rx_task_queue.TestCommonResource - Process 1 with resource-1
14:30:27.747 [RxCachedThreadScheduler-2] INFO my.test.rx_task_queue.TestCommonResource - Process 2 with resource-1
14:30:27.748 [RxCachedThreadScheduler-3] INFO my.test.rx_task_queue.TestCommonResource - Process 3 with resource-1
14:30:27.749 [RxCachedThreadScheduler-4] INFO my.test.rx_task_queue.TestCommonResource - Process 4 with resource-1
14:30:27.749 [RxCachedThreadScheduler-1] INFO my.test.rx_task_queue.TestCommonResource - Process 5 with resource-1
14:30:27.750 [main] INFO my.test.rx_task_queue.TestCommonResource - Result: [1 @ resource-1, 2 @ resource-1, 3 @ resource-1, 4 @ resource-1, 5 @ resource-1]

Использование Flowable.parallel() вместо flatMap() приводит к тому же результату.

1 Ответ

1 голос
/ 01 июля 2019

Утилизация происходит с удалением источника, поэтому, если вы хотите утилизировать после того, как все будет сделано, вам просто нужно, чтобы singleFunction возвратил весь поток: 

object TestCommonResource {
    private val logger = LoggerFactory.getLogger(TestCommonResource::class.java)
    @JvmStatic
    fun main(args: Array<String>) {
        val queue = Flowable.fromIterable(1..5)
        val resIdx = AtomicInteger(0)
        val result = Single.using({
            val res = "resource-${resIdx.incrementAndGet()}"
            logger.info("Init resource $res")
            res
        }, { res ->
            queue.flatMapSingle({ item ->
                Single.fromCallable {
                    logger.info("Process $item with $res")
                    "$item @ $res"
                }
                        .subscribeOn(Schedulers.io())
            }, false, 2)
                    .toList()
        }, { res ->
            logger.info("Dispose resource $res")
        }, false)
                .blockingGet()
        logger.info("Result: $result")
    }
}
...