使用信号量(Semaphore)实现生产者与消费者模型

Love The Way You Lie 2022-11-29 13:13 173阅读 0赞

## 什么是生产者与消费者模型？ ##

> 生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。

### 打个比方 ###

好比去肯德基吃汉堡，货架上有做好的汉堡。  
消费者花钱来消费汉堡，而肯德基的厨师则不停地做汉堡放在货架上。  
当货架满了，却没有消费者再来买汉堡时，厨师就不做了，生产者阻塞。  
当餐点来了，消费者买的太快，厨师做不过来了，货架一直空着。  
则消费者排队等着，消费者阻塞。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbWVuZ2h1YTkxMTI_size_16_color_FFFFFF_t_70_pic_center]

图：生产者-消费者模型

## 为何需要生产者-消费者模型？ ##

> 在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

## 源码实现 ##

> 假设货架上最多可以放`10个`汉堡，那么如果生产者已经生产了`10个`汉堡，却没有消费者来消费，则生产者等待。  
> 如果消费者过多，货架上一个汉堡都没有了，则消费者等待。  
> 消费者等待过程中，如果生产者又生产了汉堡放在货架上，则消费者马上消费掉这个汉堡。

需要一个生产者类:`class Producer`，用来往货架上放汉堡。  
一个消费者类：`class Consumer`，用来购买（消费）货架上的汉堡。  
一个肯德基货架类：`class KfcShelf`，包含一个`put(int hamburgerId)`方法，一个`take()`方法。 还有一个`main`方法作为程序入口。  
一个汉堡实体类，包含：`汉堡条码（id）`字段和`汉堡名称`字段。  
同时，为了控制货架上最多只能放`10`个汉堡，我们采用了`许可证数量为10`的信号量。 使用`LinkedList`作为汉堡的容器，当往货架放汉堡时，生产者用掉一个许可证，同时给消费者发放一个许可证。  
当从货架取走汉堡时，消费者用掉一个许可证，同时给生产者发放一个许可证，相当于告诉厨师：货架上空出了一个位置，赶紧补货吧。

完整源码如下：

import java.util.ArrayList;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.Semaphore;
    import java.util.stream.Collectors;
    
    import lombok.Data;
    import lombok.extern.slf4j.Slf4j;
    
    /** * <pre> * 程序目的：使用Semapore实现生产者消费者模型 * 肯德基货架的例子 * 假设货架上最多可以放10个汉堡，那么如果生产者已经生产了10个汉堡，却没有消费者来消费，则生产者等待。 * 如果消费者过多，货架上一个汉堡都没有了，则消费者等待。 * 消费者等待过程中，如果生产者又生产了汉堡放在货架上，则消费者马上消费掉这个汉堡。 * </pre> * created at 2020/8/18 09:35 * @author lerry */
    @Slf4j
    public class KfcShelf { 
    	/** * 消费者许可 */
    	Semaphore consumerPermitCount = new Semaphore(0);
    
    	/** * 生产者最多可以获取10个许可 */
    	Semaphore producerPermitCount = new Semaphore(10);
    
    	/** * 许可证数量为1、相当于普通的锁 */
    	Semaphore isUse = new Semaphore(1);
    
    	/** * 充当货架角色。注意：LinkedList是线程不安全的 */
    	LinkedList<Integer> shelfList = new LinkedList<>();
    
    	public static void main(String[] args) { 
    		KfcShelf kfcShelf = new KfcShelf();
    		// 生产者生产11个汉堡，前10个汉堡可以放在货架上，第11个生产者需要等待
    		List<Thread> producerThreadList = new ArrayList<>();
    		for (int i = 0; i < 11; i++) { 
    			Thread producer = new Thread(new Producer(kfcShelf));
    			producer.start();
    			producerThreadList.add(producer);
    		}
    
    		try { 
    			Thread.sleep(1_000);
    			// 等待消费者前来消费
    			log.info("生产10个汉堡，第11个生产者WAITING。等待消费者前来消费");
    			log.info("生产者线程状态打印：\n{}", producerThreadList.stream().map(thread -> String.format("%s:%s\n", thread.getName(), thread.getState())).collect(Collectors.joining()));
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    
    		List<Thread> consumerThreadList = new ArrayList<>();
    		// 消费者消费掉12个汉堡，第12个等待
    		for (int i = 0; i < 12; i++) { 
    			Thread consumer = new Thread(new Consumer(kfcShelf));
    			consumer.start();
    			consumerThreadList.add(consumer);
    		}
    
    		try { 
    			Thread.sleep(1_000);
    			// 等待消费者前来消费
    			log.info("消费者购买汉堡");
    			log.info("消费者线程状态打印：\n{}", consumerThreadList.stream().map(thread -> String.format("%s:%s\n", thread.getName(), thread.getState())).collect(Collectors.joining()));
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    
    
    		try { 
    			Thread.sleep(1_000);
    			// 等待生产者生产
    			log.info("等待生产者生产第12个汉堡");
    			log.info("======================");
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    		// 生产者再生产一个汉堡，则消费者马上会消费掉
    		Thread producer12 = new Thread(new Producer(kfcShelf));
    		producer12.start();
    	}
    
    	/** * 往货架上放汉堡 * @param hamburgerId */
    	public void put(int hamburgerId) { 
    		try { 
    			// 生产者获取一个许可证
    			producerPermitCount.acquire();
    			isUse.acquire();
    			shelfList.add(hamburgerId);
    			log.info("生产:[汉堡-{}] 货架上的汉堡数量:[{}]", hamburgerId, shelfList.size());
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    		finally { 
    			isUse.release();
    			consumerPermitCount.release();
    		}
    		try { 
    			Thread.sleep(1000);
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    
    	}// end method
    
    	/** * 从货架上取一个汉堡 * @return */
    	public Integer take() { 
    		Integer hamburgerId = 0;
    		try { 
    			consumerPermitCount.acquire();
    			isUse.acquire();
    			hamburgerId = shelfList.remove(0);
    			log.info("消费:[汉堡-{}] 货架上的汉堡数量:[{}]", hamburgerId, shelfList.size());
    		}
    		catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    		finally { 
    			isUse.release();
    			producerPermitCount.release();
    		} return hamburgerId;
    	}// end method
    }// end class 
    
    class Producer implements Runnable { 
    	private KfcShelf kfcShelf;
    
    	public Producer(KfcShelf KfcShelf) { 
    		this.kfcShelf = KfcShelf;
    	}
    
    	@Override
    	public void run() { 
    		// 生成6位长度的随机数，当做条码编号
    		int barCodeId = new Random().nextInt(100_000) + 100_000;
    		kfcShelf.put(barCodeId);
    	}
    }// end class
    
    class Consumer implements Runnable { 
    	private KfcShelf kfcShelf;
    
    	public Consumer(KfcShelf KfcShelf) { 
    		this.kfcShelf = KfcShelf;
    	}
    
    	@Override
    	public void run() { 
    		kfcShelf.take();
    	}
    }
    
    /** * 汉堡实体 */
    @Data
    class Hamburger { 
    	private Integer id;
    
    	private Integer name;
    }

#### 输出结果 ####

2020-08-21 16:21:39.577 [Thread-2] INFO  KfcShelf.java:109 - 生产:[汉堡-196280] 货架上的汉堡数量:[1]
    2020-08-21 16:21:39.582 [Thread-3] INFO  KfcShelf.java:109 - 生产:[汉堡-186974] 货架上的汉堡数量:[2]
    2020-08-21 16:21:39.582 [Thread-4] INFO  KfcShelf.java:109 - 生产:[汉堡-143871] 货架上的汉堡数量:[3]
    2020-08-21 16:21:39.583 [Thread-0] INFO  KfcShelf.java:109 - 生产:[汉堡-157386] 货架上的汉堡数量:[4]
    2020-08-21 16:21:39.583 [Thread-6] INFO  KfcShelf.java:109 - 生产:[汉堡-123033] 货架上的汉堡数量:[5]
    2020-08-21 16:21:39.583 [Thread-5] INFO  KfcShelf.java:109 - 生产:[汉堡-151965] 货架上的汉堡数量:[6]
    2020-08-21 16:21:39.583 [Thread-1] INFO  KfcShelf.java:109 - 生产:[汉堡-146100] 货架上的汉堡数量:[7]
    2020-08-21 16:21:39.583 [Thread-7] INFO  KfcShelf.java:109 - 生产:[汉堡-108609] 货架上的汉堡数量:[8]
    2020-08-21 16:21:39.584 [Thread-8] INFO  KfcShelf.java:109 - 生产:[汉堡-172889] 货架上的汉堡数量:[9]
    2020-08-21 16:21:39.584 [Thread-9] INFO  KfcShelf.java:109 - 生产:[汉堡-157386] 货架上的汉堡数量:[10]
    2020-08-21 16:21:40.581 [main    ] INFO  KfcShelf.java:59 - 生产10个汉堡，第11个生产者WAITING。等待消费者前来消费
    2020-08-21 16:21:40.644 [main    ] INFO  KfcShelf.java:60 - 生产者线程状态打印：
    Thread-0:TERMINATED
    Thread-1:TERMINATED
    Thread-2:TERMINATED
    Thread-3:TERMINATED
    Thread-4:TERMINATED
    Thread-5:TERMINATED
    Thread-6:TERMINATED
    Thread-7:TERMINATED
    Thread-8:TERMINATED
    Thread-9:TERMINATED
    Thread-10:WAITING
    
    2020-08-21 16:21:40.647 [Thread-11] INFO  KfcShelf.java:137 - 消费:[汉堡-196280] 货架上的汉堡数量:[9]
    2020-08-21 16:21:40.648 [Thread-16] INFO  KfcShelf.java:137 - 消费:[汉堡-186974] 货架上的汉堡数量:[8]
    2020-08-21 16:21:40.648 [Thread-12] INFO  KfcShelf.java:137 - 消费:[汉堡-143871] 货架上的汉堡数量:[7]
    2020-08-21 16:21:40.648 [Thread-13] INFO  KfcShelf.java:137 - 消费:[汉堡-157386] 货架上的汉堡数量:[6]
    2020-08-21 16:21:40.648 [Thread-14] INFO  KfcShelf.java:137 - 消费:[汉堡-123033] 货架上的汉堡数量:[5]
    2020-08-21 16:21:40.649 [Thread-15] INFO  KfcShelf.java:137 - 消费:[汉堡-151965] 货架上的汉堡数量:[4]
    2020-08-21 16:21:40.649 [Thread-10] INFO  KfcShelf.java:109 - 生产:[汉堡-147206] 货架上的汉堡数量:[5]
    2020-08-21 16:21:40.649 [Thread-17] INFO  KfcShelf.java:137 - 消费:[汉堡-146100] 货架上的汉堡数量:[4]
    2020-08-21 16:21:40.649 [Thread-18] INFO  KfcShelf.java:137 - 消费:[汉堡-108609] 货架上的汉堡数量:[3]
    2020-08-21 16:21:40.649 [Thread-19] INFO  KfcShelf.java:137 - 消费:[汉堡-172889] 货架上的汉堡数量:[2]
    2020-08-21 16:21:40.650 [Thread-20] INFO  KfcShelf.java:137 - 消费:[汉堡-157386] 货架上的汉堡数量:[1]
    2020-08-21 16:21:40.650 [Thread-21] INFO  KfcShelf.java:137 - 消费:[汉堡-147206] 货架上的汉堡数量:[0]
    2020-08-21 16:21:41.652 [main    ] INFO  KfcShelf.java:77 - 消费者购买汉堡
    2020-08-21 16:21:41.655 [main    ] INFO  KfcShelf.java:78 - 消费者线程状态打印：
    Thread-11:TERMINATED
    Thread-12:TERMINATED
    Thread-13:TERMINATED
    Thread-14:TERMINATED
    Thread-15:TERMINATED
    Thread-16:TERMINATED
    Thread-17:TERMINATED
    Thread-18:TERMINATED
    Thread-19:TERMINATED
    Thread-20:TERMINATED
    Thread-21:TERMINATED
    Thread-22:WAITING
    
    2020-08-21 16:21:42.659 [main    ] INFO  KfcShelf.java:88 - 等待生产者生产第12个汉堡
    2020-08-21 16:21:42.659 [main    ] INFO  KfcShelf.java:89 - ======================
    2020-08-21 16:21:42.660 [Thread-23] INFO  KfcShelf.java:109 - 生产:[汉堡-100422] 货架上的汉堡数量:[1]
    2020-08-21 16:21:42.660 [Thread-22] INFO  KfcShelf.java:137 - 消费:[汉堡-100422] 货架上的汉堡数量:[0]

#### 结果分析 ####

刚开始，货架为空。 这时，创建了`11`个生产者线程，往货架上放汉堡。

for (int i = 0; i < 11; i++) { 
    	Thread producer = new Thread(new Producer(kfcShelf));
    	producer.start();
    	producerThreadList.add(producer);
    }

可是，生产者的许可证数量为`10`，第`11个生产者`获取不到许可证，怎么办？ 线程等待。

Thread-10:WAITING

生产者的线程状态都打印出来后，可以看到，第`11个生产者`线程状态为`WAITING`。  
查看`AQS`源码，可以发现，当线程获取不到许可证时，会：

private final boolean parkAndCheckInterrupt() { 
        LockSupport.park(this);
        return Thread.interrupted();
    }

调用`LockSupport.park`方法。然后线程进入`WAITING`状态。

然后，来了`12`个消费者：

for (int i = 0; i < 12; i++) { 
    	Thread consumer = new Thread(new Consumer(kfcShelf));
    	consumer.start();
    	consumerThreadList.add(consumer);
    }

货架上的`10`个汉堡，被取了之后，等待中的那个生产者线程会被激活：

2020-08-21 16:21:40.649 [Thread-10] INFO  KfcShelf.java:109 - 生产:[汉堡-147206] 货架上的汉堡数量:[5]

它生产了编号为`147206`的汉堡。随后，被消费者线程`Thread-21`消费掉。

2020-08-21 16:21:40.650 [Thread-21] INFO  KfcShelf.java:137 - 消费:[汉堡-147206] 货架上的汉堡数量:[0]

这时，生产的`11`个汉堡，都被消费掉了。第`12`个消费者没有汉堡买了，怎么办呢？ 等待。

Thread-22:WAITING

然后，我们启动第`12`个生产者线程：

// 生产者再生产一个汉堡，则消费者马上会消费掉
    Thread producer12 = new Thread(new Producer(kfcShelf));
    producer12.start();

看日志可以发现，生产出来的编号`100422`的汉堡，马上被第`12`个消费者消费掉了。

2020-08-21 16:21:42.660 [Thread-23] INFO  KfcShelf.java:109 - 生产:[汉堡-100422] 货架上的汉堡数量:[1]
    2020-08-21 16:21:42.660 [Thread-22] INFO  KfcShelf.java:137 - 消费:[汉堡-100422] 货架上的汉堡数量:[0]

## 参考资料 ##

[生产者与消费者模型][Link 1]  
[Java实现生产者和消费者的5种方式][Java_5]  
[经典并发同步模式：生产者-消费者设计模式 - 知乎][-_ -]

# 环境说明 #

*  java -version

java version "1.8.0_251"
    Java(TM) SE Runtime Environment (build 1.8.0_251-b08)
    Java HotSpot(TM) 64-Bit Server VM (build 25.251-b08, mixed mode)

*  OS：`macOS High Sierra 10.13.4`
 *  日志：`logback`

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbWVuZ2h1YTkxMTI_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/11fc23f3e88c4bb2b626985beb2326a5.png
[Link 1]: https://www.jianshu.com/p/f53fb95b5820
[Java_5]: https://juejin.im/entry/6844903486895865864
[-_ -]: https://zhuanlan.zhihu.com/p/73442055