服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - 孔乙己:Kotlin生产者消费者问题的八种解法

孔乙己:Kotlin生产者消费者问题的八种解法

2021-08-31 22:57AndroidPubfundroid Java教程

生产者和消费者问题是线程模型中的经典问题:生产者和消费者在同一时间段内共用同一个缓冲区(Buffer),生产者往 Buffer 中添加产品,消费者从 Buffer 中取走产品,当 Buffer 为空时,消费者阻塞,当 Buffer 满时,生产者阻塞。

孔乙己:Kotlin生产者消费者问题的八种解法

孔乙己:Kotlin生产者消费者问题的八种解法

生产者和消费者问题是线程模型中的经典问题:生产者和消费者在同一时间段内共用同一个缓冲区(Buffer),生产者往 Buffer 中添加产品,消费者从 Buffer 中取走产品,当 Buffer 为空时,消费者阻塞,当 Buffer 满时,生产者阻塞。

Kotlin 中有多种方法可以实现多线程的生产/消费模型(大多也适用于Java)

  1. Synchronized
  2. ReentrantLock
  3. BlockingQueue
  4. Semaphore
  5. PipedXXXStream
  6. RxJava
  7. Coroutine
  8. Flow

1. Synchronized

Synchronized 是最最基本的线程同步工具,配合 wait/notify 可以实现实现生产消费问题。

  1. val buffer = LinkedList<Data>() 
  2. val MAX = 5 //buffer最大size 
  3.  
  4. val lock = Object() 
  5.  
  6. fun produce(data: Data) { 
  7.     sleep(2000) // mock produce 
  8.     synchronized(lock) { 
  9.         while (buffer.size >= MAX) { 
  10.            // 当buffer满时,停止生产 
  11.            // 注意此处使用while不能使用if,因为有可能是被另一个生产线程而非消费线程唤醒,所以要再次检查buffer状态 
  12.            // 如果生产消费两把锁,则不必担心此问题 
  13.            lock.wait() 
  14.         } 
  15.  
  16.       buffer.push(data) 
  17.         // notify方法只唤醒其中一个线程,选择哪个线程取决于操作系统对多线程管理的实现。 
  18.         // notifyAll会唤醒所有等待中线程,哪一个线程将会第一个处理取决于操作系统的实现,但是都有机会处理。 
  19.         // 此处使用notify有可能唤醒的是另一个生产线程从而造成死锁,所以必须使用notifyAll 
  20.         lock.notifyAll() 
  21.     } 
  22.  
  23. fun consume() { 
  24.     synchronized(lock) { 
  25.         while (buffer.isEmpty()) 
  26.             lock.wait() // 暂停消费 
  27.         buffer.removeFirst() 
  28.         lock.notifyAll() 
  29.     } 
  30.     sleep(2000) // mock consume 
  31.  
  32.  
  33.  
  34. @Test 
  35. fun test() { 
  36.     // 同时启动多个生产、消费线程 
  37.     repeat(10) { 
  38.         Thread { produce(Data()) }.start() 
  39.     } 
  40.     repeat(10) { 
  41.         Thread { consume() }.start() 
  42.     } 

2. ReentrantLock

Lock 相对于 Synchronized 好处是当有多个生产线/消费线程时,我们可以通过定义多个 condition 精确指定唤醒哪一个。下面的例子展示 Lock 配合 await/single 替换前面 Synchronized 写法。

  1. val buffer = LinkedList<Data>() 
  2. val MAX = 5 //buffer最大size 
  3.               
  4. val lock = ReentrantLock()                      
  5. val condition = lock.newCondition()           
  6.                                                 
  7. fun produce(data: Data) {                       
  8.     sleep(2000) // mock produce                 
  9.     lock.lock()                                 
  10.                                                 
  11.     while (buffer.size >= 5)                       
  12.         condition.await()                       
  13.                                                 
  14.     buffer.push(data)                           
  15.     condition.signalAll()                       
  16.     lock.unlock()                               
  17. }                                               
  18.                                                 
  19. fun consume() {                                 
  20.     lock.lock()                                 
  21.     while (buffer.isEmpty())                       
  22.         condition.await()                       
  23.                                                 
  24.     buffer.removeFirst() 
  25.     condition.singleAll()                         
  26.     lock.unlock()                               
  27.     sleep(2000) // mock consume                 
  28. }                                             

3. BlockingQueue (阻塞队列)

BlockingQueue在达到临界条件时,再进行读写会自动阻塞当前线程等待锁的释放,天然适合这种生产/消费场景。

  1. val buffer = LinkedBlockingQueue<Data>(5)                
  2.                                                          
  3. fun produce(data: Data) {                                
  4.     sleep(2000) // mock produce                          
  5.     buffer.put(data) //buffer满时自动阻塞                        
  6.                                                          
  7. fun consume() {                                          
  8.     buffer.take() // buffer空时自动阻塞 
  9.     sleep(2000) // mock consume                          
  10. }                                                        
  11.                                       

注意 BlockingQueue 的有三组读/写方法,只有一组有阻塞效果,不要用错。

方法 说明
add(o)/remove(o) add 方法在添加元素的时候,若超出了队列的长度会直接抛出异常
offer(o)/poll(o) offer 在添加元素时,如果发现队列已满无法添加的话,会直接返回false
put(o)/take(o) put 向队尾添加元素的时候发现队列已经满了会发生阻塞一直等待空间,以加入元素
 

4. Semaphore(信号量)

Semaphore 是 JUC 提供的一种共享锁机制,可以进行拥塞控制,此特性可用来控制 buffer 的大小。

  1. // canProduce: 可以生产数量(即buffer可用的数量),生产者调用acquire,减少permit数目     
  2. val canProduce = Semaphore(5)                                                                                            
  3. // canConsumer:可以消费数量,生产者调用release,增加permit数目                   
  4. val canConsume = Semaphore(5)                                                                                       
  5. // 控制buffer访问互斥                                                 
  6. val mutex = Semaphore(0)                                        
  7.                                                                 
  8. val buffer = LinkedList<Data>()                                 
  9.                                                                 
  10. fun produce(data: Data) {                                       
  11.     if (canProduce.tryAcquire()) {                              
  12.         sleep(2000) // mock produce                             
  13.                                                                 
  14.         mutex.acquire()                                         
  15.         buffer.push(data)                                       
  16.         mutex.release()                                         
  17.                                                                 
  18.         canConsume.release() //通知消费端新增加了一个产品                    
  19.     }                                                           
  20. }                                                               
  21.                                                                 
  22. fun consume() {                                                 
  23.     if (canConsume.tryAcquire()) {                              
  24.         sleep(2000) // mock consume                             
  25.                                                                 
  26.         mutex.acquire()                                         
  27.         buffer.removeFirst()                                    
  28.         mutex.release()                                         
  29.                                                                 
  30.         canProduce.release() //通知生产端可以再追加生产                     
  31.     }                                                           
  32.                                                                 
  33. }                                         

5. PipedXXXStream (管道)

Java 里的管道输入/输出流 PipedInputStream / PipedOutputStream 实现了类似管道的功能,用于不同线程之间的相互通信,输入流中有一个缓冲数组,当缓冲数组为空的时候,输入流 PipedInputStream 所在的线程将阻塞。

  1. val pis: PipedInputStream = PipedInputStream() 
  2. val pos: PipedOutputStream by lazy { 
  3.     PipedOutputStream().apply { 
  4.         pis.connect(this) //输入输出流之间建立连接 
  5.     } 
  6.  
  7. fun produce(data: ContactsContract.Data) { 
  8.     while (true) { 
  9.         sleep(2000) 
  10.         pos.use { // Kotlin 使用 use 方便的进行资源释放 
  11.             it.write(data.getBytes()) 
  12.             it.flush() 
  13.         } 
  14.     } 
  15.  
  16. fun consume() { 
  17.     while (true) { 
  18.         sleep(2000) 
  19.         pis.use { 
  20.             val byteArray = ByteArray(1024) 
  21.             it.read(byteArray) 
  22.         } 
  23.     } 
  24.  
  25. @Test 
  26. fun Test() { 
  27.     repeat(10) { 
  28.         Thread { produce(Data()) }.start() 
  29.     } 
  30.  
  31.     repeat(10) { 
  32.         Thread { consume() }.start() 
  33.     } 

6. RxJava

RxJava 从概念上,可以将 Observable/Subject 作为生产者, Subscriber 作为消费者, 但是无论 Subject 或是 Observable 都缺少 Buffer 溢出时的阻塞机制,难以独立实现生产者/消费者模型。

Flowable 的背压机制,可以用来控制 buffer 数量,并在上下游之间建立通信, 配合 Atomic 可以变向实现单生产者/单消费者场景,(不适用于多生产者/多消费者场景)。

  1. class Producer : Flowable<Data>() { 
  2.  
  3.     override fun subscribeActual(subscriber: org.reactivestreams.Subscriber<in Data>) { 
  4.         subscriber.onSubscribe(object : Subscription { 
  5.             override fun cancel() { 
  6.                 //... 
  7.             } 
  8.  
  9.             private val outStandingRequests = AtomicLong(0) 
  10.  
  11.             override fun request(n: Long) { //收到下游通知,开始生产 
  12.                 outStandingRequests.addAndGet(n) 
  13.  
  14.                 while (outStandingRequests.get() > 0) { 
  15.                     sleep(2000) 
  16.                     subscriber.onNext(Data()) 
  17.                     outStandingRequests.decrementAndGet() 
  18.                 } 
  19.             } 
  20.  
  21.         }) 
  22.     } 
  23.  
  24.  
  25.  
  26.  
  27. class Consumer : DefaultSubscriber<Data>() { 
  28.  
  29.     override fun onStart() { 
  30.         request(1) 
  31.     } 
  32.  
  33.     override fun onNext(i: Data?) { 
  34.         sleep(2000) //mock consume 
  35.         request(1) //通知上游可以增加生产 
  36.     } 
  37.  
  38.     override fun onError(throwable: Throwable) { 
  39.         //... 
  40.     } 
  41.  
  42.     override fun onComplete() { 
  43.         //... 
  44.     } 
  45.  
  46.  
  47.  
  48. @Test 
  49. fun test_rxjava() { 
  50.     try { 
  51.         val testProducer = Producer) 
  52.         val testConsumer = Consumer() 
  53.  
  54.         testProducer 
  55.             .subscribeOn(Schedulers.computation()) 
  56.             .observeOn(Schedulers.single()) 
  57.             .blockingSubscribe(testConsumer) 
  58.  
  59.     } catch (t: Throwable) { 
  60.         t.printStackTrace() 
  61.     } 
  62.  

7. Coroutine Channel

协程中的 Channel 具有拥塞控制机制,可以实现生产者消费者之间的通信。可以把 Channel 理解为一个协程版本的阻塞队列,capacity 指定队列容量。

  1. val channel = Channel<Data>(capacity = 5) 
  2.  
  3. suspend fun produce(data: ContactsContract.Contacts.Data) = run { 
  4.     delay(2000) //mock produce 
  5.     channel.send(data) 
  6.  
  7.  
  8. suspend fun consume() = run { 
  9.     delay(2000)//mock consume 
  10.     channel.receive() 
  11.  
  12. @Test 
  13. fun test_channel() { 
  14.     repeat(10) { 
  15.         GlobalScope.launch { 
  16.             produce(Data()) 
  17.         } 
  18.     } 
  19.  
  20.     repeat(10) { 
  21.         GlobalScope.launch { 
  22.            consume() 
  23.         } 
  24.     } 

此外,Coroutine 提供了 produce 方法,在声明 Channel 的同时生产数据,写法上更简单,适合单消费者单生产者的场景:

  1. fun CoroutineScope.produce(): ReceiveChannel<Data> = produce { 
  2.     repeat(10) { 
  3.         delay(2000) //mock produce 
  4.         send(Data()) 
  5.     } 
  6.  
  7. @Test 
  8. fun test_produce() { 
  9.     GlobalScope.launch { 
  10.         produce.consumeEach { 
  11.             delay(2000) //mock consume 
  12.         } 
  13.     } 

8. Coroutine Flow

Flow 跟 RxJava 一样,因为缺少 Buffer 溢出时的阻塞机制,不适合处理生产消费问题,其背压机制也比较简单,无法像 RxJava 那样收到下游通知。但是 Flow 后来发布了 SharedFlow, 作为带缓冲的热流,提供了 Buffer 溢出策略,可以用作生产者/消费者之间的同步。

  1. val flow : MutableSharedFlow<Data> = MutableSharedFlow( 
  2.     extraBufferCapacity = 5  //缓冲大小 
  3.     , onBufferOverflow = BufferOverflow.SUSPEND // 缓冲溢出时的策略:挂起 
  4.  
  5. @Test 
  6. fun test() { 
  7.  
  8.     GlobalScope.launch { 
  9.         repeat(10) { 
  10.             delay(2000) //mock produce 
  11.             sharedFlow.emit(Data()) 
  12.         } 
  13.     } 
  14.  
  15.     GlobalScope.launch { 
  16.         sharedFlow.collect { 
  17.             delay(2000) //mock consume 
  18.         } 
  19.     } 

注意 SharedFlow 也只能用在单生产者/单消费者场景。

总结

生产者/消费者问题,其本质核心还是多线程读写共享资源(Buffer)时的同步问题,理论上只要具有同步机制的多线程框架,例如线程锁、信号量、阻塞队列、协程 Channel等,都是可以实现生产消费模型的。

另外,RxJava 和 Flow 虽然也是多线程框架,但是缺少Buffer溢出时的阻塞机制,不适用于生产/消费场景,更适合在纯响应式场景中使用。

原文链接:https://mp.weixin.qq.com/s/zWrSq8JVRkLUilMMgdv1Cw

延伸 · 阅读

精彩推荐