RxJava BackPressureStrategy.DROP排队新任务，而不是放弃它们(RxJava BackPressureStrategy.DROP queues new tasks instead of dropping them)

 我的Android活动中有以下代码，可以按预期工作。 在滚动事件中，它通过带有SynchronousQueue ＆ DiscardPolicy的ThreadPoolExecutor调用expensiveHttpCall() ，这会在最后一次调用仍在运行时丢弃新的调用。 
 
class MyActivity : AppCompatActivity() {

    val recyclerView: RecyclerView
.
    val oneAtATimeExecutor = ThreadPoolExecutor(1, 1, 1L, TimeUnit.SECONDS,
            SynchronousQueue<Runnable>(),
            ThreadPoolExecutor.DiscardPolicy()
    )

    private val scrollListener: RecyclerView.OnScrollListener = object : RecyclerView.OnScrollListener() {
        override fun onScrolled(recyclerView: RecyclerView?, dx: Int, dy: Int) {
            super.onScrolled(recyclerView, dx, dy)
            Log.e("rx","triggered!")
            oneAtATimeExecutor.execute { expensiveHttpCall() }
        }
    }

    override fun onCreate(savedInstanceState: Bundle?) {

        //...

        recyclerView.addOnScrollListener(scrollListener)

        //...
    }

    private fun expensiveHttpCall() {
        Thread.sleep(2000L)
        Log.e("rx", "done!")
    }
}
 
 我在使用RxJava的PublishProcessor遇到了同样的PublishProcessor ： 
 
class MyActivity : AppCompatActivity() {

    val recyclerView: RecyclerView

    val processor = PublishProcessor.create<Unit>()

    private val scrollListener: RecyclerView.OnScrollListener = object : RecyclerView.OnScrollListener() {
        override fun onScrolled(recyclerView: RecyclerView?, dx: Int, dy: Int) {
            super.onScrolled(recyclerView, dx, dy)
            Log.e("rx","triggered!")
            processor.onNext(Unit)
        }
    }

    override fun onCreate(savedInstanceState: Bundle?) {

        //...

        recyclerView.addOnScrollListener(scrollListener)

        //...

        processor.onBackpressureDrop().observeOn(Schedulers.newThread())
                .subscribe { expensiveHttpCall() }
    }

    private fun expensiveHttpCall() {
        Thread.sleep(2000L)
        Log.e("rx", "done!")
    }
}
 
 令人惊讶的滚动后，控制台打印出一堆"triggered"消息。 然后以2秒的连续速率打印"done!" 多达"triggered"打印的次数。 这有效地表现得像一个队列，而不像预期的那样。 我究竟做错了什么？ 

I have the code below in my Android activity which works as intended. on scroll events it invokes the expensiveHttpCall() through the ThreadPoolExecutor with SynchronousQueue & DiscardPolicy which results on new invokations to be discarded while the last invokation is still running.
 
class MyActivity : AppCompatActivity() {

    val recyclerView: RecyclerView
.
    val oneAtATimeExecutor = ThreadPoolExecutor(1, 1, 1L, TimeUnit.SECONDS,
            SynchronousQueue<Runnable>(),
            ThreadPoolExecutor.DiscardPolicy()
    )

    private val scrollListener: RecyclerView.OnScrollListener = object : RecyclerView.OnScrollListener() {
        override fun onScrolled(recyclerView: RecyclerView?, dx: Int, dy: Int) {
            super.onScrolled(recyclerView, dx, dy)
            Log.e("rx","triggered!")
            oneAtATimeExecutor.execute { expensiveHttpCall() }
        }
    }

    override fun onCreate(savedInstanceState: Bundle?) {

        //...

        recyclerView.addOnScrollListener(scrollListener)

        //...
    }

    private fun expensiveHttpCall() {
        Thread.sleep(2000L)
        Log.e("rx", "done!")
    }
}
 
I'm having trouble implementing the same behavior with RxJava's PublishProcessor:
 
class MyActivity : AppCompatActivity() {

    val recyclerView: RecyclerView

    val processor = PublishProcessor.create<Unit>()

    private val scrollListener: RecyclerView.OnScrollListener = object : RecyclerView.OnScrollListener() {
        override fun onScrolled(recyclerView: RecyclerView?, dx: Int, dy: Int) {
            super.onScrolled(recyclerView, dx, dy)
            Log.e("rx","triggered!")
            processor.onNext(Unit)
        }
    }

    override fun onCreate(savedInstanceState: Bundle?) {

        //...

        recyclerView.addOnScrollListener(scrollListener)

        //...

        processor.onBackpressureDrop().observeOn(Schedulers.newThread())
                .subscribe { expensiveHttpCall() }
    }

    private fun expensiveHttpCall() {
        Thread.sleep(2000L)
        Log.e("rx", "done!")
    }
}
 
To my surprise after a scroll the console prints out a bunch of "triggered" messages. then on a 2 second succession rate it prints "done!" for as many times as the number of "triggered" prints. this effectively behaves like a queue rather than as was intended. what am I doing wrong?

原文：https://stackoverflow.com/questions/50084119