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

 

目前成都创新互联已为数千家的企业提供了网站建设、域名、网页空间、绵阳服务器托管、企业网站设计、焉耆网站维护等服务，公司将坚持客户导向、应用为本的策略，正道将秉承"和谐、参与、激情"的文化，与客户和合作伙伴齐心协力一起成长，共同发展。

本文转载自微信公众号「AndroidPub」，作者fundroid。转载本文请联系AndroidPub公众号。

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

2. ReentrantLock

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

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

3. BlockingQueue (阻塞队列)

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

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

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

5. PipedXXXStream (管道)

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

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

6. RxJava

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

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

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

7. Coroutine Channel

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

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

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

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

8. Coroutine Flow

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

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

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

总结

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

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

本文标题：孔乙己：Kotlin生产者消费者问题的八种解法
文章分享：http://www.shufengxianlan.com/qtweb/news45/522095.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

微信小程序知识

各行业网站

响应式网站设计　　 　资阳服务器托管　　 　企业网站维护　　 　成都微信开发　　 　成都花箱厂家　　 　云主机　　 　成都广告公司　　 　成都品牌网站建设　　 　网络推广外包　　 　网络推广公司　　 　网络推广　　 　四川审计公司　　 　兴发搬家　　 　复盛机电设备　　 　专业网站建设　　 　网站制作价格　　 　绵阳机房托管　　 　东方电机技改　　 　响应式网站建设方案　　 　酒柜书柜鞋柜定制