线程池+阻塞队列 模仿生产者消费者
线程池+阻塞队列 模仿生产者消费者
lintalkliu 发表于2年前
线程池+阻塞队列 模仿生产者消费者
  • 发表于 2年前
  • 阅读 81
  • 收藏 1
  • 点赞 2
  • 评论 0

移动开发云端新模式探索实践 >>>   

入口点

[java] view plain copy

  1. package multithreading.BlockingQueue;  
  2.   
  3. import java.util.concurrent.BlockingQueue;   
  4. import java.util.concurrent.ExecutorService;   
  5. import java.util.concurrent.Executors;   
  6. import java.util.concurrent.LinkedBlockingQueue;   
  7. import java.util.concurrent.TimeUnit;  
  8.     
  9. /**  
  10.  * Throws exception    Specical value        TimeOut                    Block 
  11.  * add(e)                       offer(e)               offer(e,time,unit)         put 
  12.  * remove                      poll                   poll(time,unit)             take 
  13.  */  
  14. public class BlockingQueueTest {   
  15.     
  16.     public static void main(String[] args) throws InterruptedException {   
  17.         // 声明一个容量为10的缓存队列   
  18.         //SynchronousQueue<String> 只有有人来拿的时候数据才能放的进去  
  19.         BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);   
  20.           
  21.         Producer producer1 = new Producer("rain",queue);   
  22.         Producer producer2 = new Producer("tom",queue);   
  23.         Producer producer3 = new Producer("jack",queue);   
  24.         Consumer consumer = new Consumer(queue);   
  25.     
  26.         // 借助Executors 线程池  
  27.         ExecutorService service = Executors.newCachedThreadPool();   
  28.         // 启动线程   
  29.         service.execute(producer1);   
  30.         service.execute(producer2);   
  31.         service.execute(producer3);   
  32.         service.execute(consumer);   
  33.     
  34.         // 执行3s   
  35.         TimeUnit.SECONDS.sleep(3);  
  36.         producer1.stop();   
  37.         producer2.stop();   
  38.         producer3.stop();   
  39.     
  40.         // 退出Executor   
  41.         service.shutdown();   
  42.     }   
  43. }   


 

 

生产者

 

[java] view plain copy

  1. package multithreading.BlockingQueue;  
  2.   
  3. import java.util.Random;   
  4. import java.util.concurrent.BlockingQueue;   
  5. import java.util.concurrent.TimeUnit;   
  6. import java.util.concurrent.atomic.AtomicInteger;  
  7.   
  8. public class Producer implements Runnable {   
  9.     
  10.     private String name;  
  11.     private volatile boolean      isRunning               = true;   
  12.     private BlockingQueue queue;   
  13.     private static AtomicInteger  count                   = new AtomicInteger();   
  14.     private static final int      DEFAULT_RANGE_FOR_SLEEP = 1000;   
  15.       
  16.     public Producer(String name,BlockingQueue queue) {   
  17.         this.queue = queue;   
  18.         this.name=name;  
  19.     }   
  20.     
  21.     public void run() {   
  22.         String data = null;   
  23.         Random r = new Random();   
  24.     
  25.         System.out.println("启动生产者线程!"+name);   
  26.         try {   
  27.             while (isRunning) {   
  28.                 System.out.println(name+"正在生产数据...");   
  29.                 Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));   
  30.     
  31.                 data = "data:" + count.incrementAndGet();   
  32.                 System.out.println(name+"将数据:" + data + "放入队列...");   
  33.                 if (!queue.offer(data, 2, TimeUnit.SECONDS)) {   
  34.                     System.out.println(name+"放入数据失败:" + data);   
  35.                 }   
  36.             }   
  37.         } catch (InterruptedException e) {   
  38.             e.printStackTrace();   
  39.             Thread.currentThread().interrupt();   
  40.         } finally {   
  41.             System.out.println(name+"退出生产者线程!");   
  42.         }   
  43.     }   
  44.     
  45.     public void stop() {   
  46.         isRunning = false;   
  47.     }   
  48.     
  49.   
  50.     
  51. }   


 

消费者

 

[java] view plain copy

  1. package multithreading.BlockingQueue;  
  2.   
  3. import java.util.Random;   
  4. import java.util.concurrent.BlockingQueue;   
  5. import java.util.concurrent.TimeUnit;   
  6.   
  7.   
  8. public class Consumer implements Runnable {   
  9.     
  10.         
  11.     private BlockingQueue<String> queue;   
  12.     private static final int      DEFAULT_RANGE_FOR_SLEEP = 1000;   
  13.       
  14.     public Consumer(BlockingQueue<String> queue) {   
  15.         this.queue = queue;   
  16.     }   
  17.     
  18.     public void run() {   
  19.         System.out.println("启动消费者线程!");   
  20.         Random r = new Random();   
  21.         boolean isRunning = true;   
  22.         try {   
  23.             while (isRunning) {   
  24.                 System.out.println("消费者正从队列获取数据...");   
  25.                 String data = queue.poll(5, TimeUnit.SECONDS);   
  26.                 if (null != data) {   
  27.                     System.out.println("消费者拿到数据:" + data);   
  28.                     System.out.println("消费者正在消费数据:" + data);   
  29.                     Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));   
  30.                 } else {   
  31.                     // 超过5s还没数据,认为所有生产线程都已经退出,自动退出消费线程。   
  32.                     isRunning = false;   
  33.                 }   
  34.             }   
  35.         } catch (InterruptedException e) {   
  36.             e.printStackTrace();   
  37.             Thread.currentThread().interrupt();   
  38.         } finally {   
  39.             System.out.println("退出消费者线程!");   
  40.         }   
  41.     }   
  42.   
  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
粉丝 0
博文 5
码字总数 729
×
lintalkliu
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: