ThreadPoolExecutor线程池方式实现生产者消费者模式

「爱情、让人受尽委屈。」 2022-04-04 09:08 362阅读 0赞

# 实现多线程的生产者消费者模式。 #

## **描述**：利用ThreadPoolExecutor线程池，LinkedBlockingQueue队列进行消息缓存，实现生产者消费者模式。 ##

### 本次实验难点(对本人菜鸟而言)： ###

1.  使用计时的手段来判断是否已经没有生产者提供消息了，以便关闭消费者线程。
2.  第一次使用LinkBlockingQueue的方式实现,所以没有区分add(),put的区别,以及take(),poll()的区别。导致前期没办法进入计时的代码，程序一直没有关闭。
3.  程序没有关闭的原因还有是线程池没有手动shutdown()的缘故，放到main()后面，就行了。在main方法结尾必须判断是否还有存活的线程，然后手动关闭线程池，这样才可以让程序自动停止。

--------------------

BlockingQueue定义的常用方法如下

<table> 
 <tbody> 
 <tr> 
 <td> &nbsp; </td> 
 <td> 抛出异常 </td> 
 <td> 特殊值 </td> 
 <td> 阻塞 </td> 
 <td> 超时 </td> 
 </tr> 
 <tr> 
 <td> 插入 </td> 
 <td> add(e) </td> 
 <td> offer(e) </td> 
 <td> put(e) </td> 
 <td> offer(e,time,unit) </td> 
 </tr> 
 <tr> 
 <td> 移除 </td> 
 <td> remove() </td> 
 <td> poll() </td> 
 <td> take() </td> 
 <td> poll(time,unit) </td> 
 </tr> 
 <tr> 
 <td> 检查 </td> 
 <td> element() </td> 
 <td> peek() </td> 
 <td> 不可用 </td> 
 <td> 不可用 </td> 
 </tr> 
 </tbody> 
</table>

1)add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则招聘异常

2)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.

3)put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.

4)poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null

5)take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止

其中：BlockingQueue 不接受null 元素。试图add、put 或offer 一个null 元素时，某些实现会抛出NullPointerException。null 被用作指示poll 操作失败的警戒值。

--------------------

具体实现：

--------------------

package com.example.myaop.provider;
    
    import java.time.LocalDateTime;
    import java.time.format.DateTimeFormatter;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    /**
    *
    *生产者
    */
    public class MyProvider implements Runnable {
       private volatile LinkedBlockingQueue queue;
       private static boolean flag = false;
    
       public MyProvider(LinkedBlockingQueue queue) {
          this.queue = queue;
       }
    
       public void toStop(){
          System.out.println("stop..");
          this.flag=true;
       }
    
       @Override
       public void run() {
          while (!flag){
             try {
                queue.put(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))+"mytime");
                queue.put("mytime");
             } catch (InterruptedException e) {
                e.printStackTrace();
             }
    
             try {
                System.out.println(Thread.currentThread().getName()+" sleep 500ms");
                Thread.sleep(500);
             } catch (InterruptedException e) {
                e.printStackTrace();
             }
          }
       }
    }

2.消费者：

package com.example.myaop.provider;
    
    import org.springframework.util.StringUtils;
    
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class MyConsumer implements Runnable{
          private volatile LinkedBlockingQueue queue;
          private static boolean stopFlag=false;
    
       public MyConsumer(LinkedBlockingQueue queue) {
          this.queue = queue;
       }
    
    
       @Override
       public void run() {
          System.out.println("consumer:"+Thread.currentThread().getName()+" running");
          int count = 0;
          while(!stopFlag){
             try {
                Object take = queue.poll();
                if(take != null){
                   count =0;
                   System.out.println("consumer:"+take);
                   Thread.sleep(1000);
                }else{
                   count++;
                   Thread.sleep(500);
                   System.out.println("consumer waiting");
                }
                if(count >= 20){
                   System.out.println(Thread.currentThread().getName()+"consumer stop");
                   stopFlag=true;
                }
             } catch (InterruptedException e) {
                e.printStackTrace();
             }
    
          }
       }
    }

3.Main

package com.example.myaop.provider;
 
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 public class Main {
 public static void main(String[] args) throws InterruptedException {
 ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 8, 2, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000));
 LinkedBlockingQueue queue = new LinkedBlockingQueue(3);
 MyConsumer myConsumer = new MyConsumer(queue);
 MyProvider myProvider = new MyProvider(queue);
 MyConsumer myConsumer2 = new MyConsumer(queue);
 MyProvider myProvider2 = new MyProvider(queue);
 executor.execute(myProvider);
 executor.execute(myProvider2);
 executor.execute(myConsumer);
 executor.execute(myConsumer2);
 
 Thread.sleep(3000);
 myProvider.toStop();
 myProvider2.toStop();
 while(true){
 Thread.sleep(1000);
 int activeCount = executor.getActiveCount();
 System.out.println("alive thread count:"+activeCount);
 if(activeCount == 0){
 System.out.println("executor shutdown");
 executor.shutdown();
 break;
 }
 }
 }
 }