• 多线程编程模式-Producer-consumer

    5.生产者消费者Producer-consumer

    避免生产和消费速率的差异，引入一个channel，对二者解耦

    实现producer的时候用到了两阶段模式的AbstractTerminatedThread：

    Channel.java
    1. package comsumerproducer;
    2.  
    3. /**
    4.  * Channel
    5.  */
    6. public interface Channel
       {
    7.     /**
    8.      * 取出一个产品
    9.      * @return
    10.      * @throws InterruptedException
    11.      */
    12.     P take() throws InterruptedException;
    13.  
    14.     /**
    15.      * 生产一个产品放入通道
    16.      * @param product
    17.      * @throws InterruptedException
    18.      */
    19.     void put(P product) throws InterruptedException;
    20. }
    21.  
    BlockingQueueChannel implements Channel
    1. package comsumerproducer;
    2.  
    3. import java.util.concurrent.BlockingQueue;
    4.  
    5. /**
    6.  * 基于阻塞队列的通道实现
    7.  */
    8. public class BlockingQueueChannel
       implements Channel
       {
    9.     private final BlockingQueue
       queue;
    10.  
    11.     public BlockingQueueChannel(BlockingQueue
       queue) {
    12.         this.queue = queue;
    13.     }
    14.  
    15.     @Override
    16.     public P take() throws InterruptedException {
    17.         return queue.take();
    18.     }
    19.  
    20.     @Override
    21.     public void put(P product) throws InterruptedException {
    22.         queue.put(product);
    23.     }
    24. }
    25.  
    AttachmentProcessor 角色Producer

    1. package comsumerproducer;
    2.  
    3. import twophased.AbstractTerminatedThread;
    4.  
    5. import java.io.*;
    6. import java.text.Normalizer;
    7. import java.util.Random;
    8. import java.util.concurrent.ArrayBlockingQueue;
    9.  
    10. /**
    11.  * 模式角色：producer
    12.  */
    13. public class AttachmentProcessor {
    14.     private final String ATTACHMENT_STORE_BASE_DIR =
    15.             "/home/viscent/tmp/attachments/";
    16.  
    17.     /**
    18.      * 模式角色：Channel
    19.      */
    20.     private final Channel channel =
    21.             new BlockingQueueChannel(new ArrayBlockingQueue(200));
    22.  
    23.     /**
    24.      * 模式角色：Consumer
    25.      */
    26.     private final AbstractTerminatedThread indexingThread = new AbstractTerminatedThread() {
    27.         @Override
    28.         protected void doRun() throws Exception {
    29.             File file = null;
    30.             file = channel.take();
    31.             try {
    32.                 indexFile(file);
    33.             } catch (Exception e) {
    34.                 e.printStackTrace();
    35.             } finally {
    36.                 terminationToken.reservations.decrementAndGet();
    37.             }
    38.         }
    39.  
    40.         private void indexFile(File file) throws Exception {
    41.             /**
    42.              * 省略其他代码
    43.               */
    44.             /**
    45.              * 模拟产生索引文件的时间消耗
    46.              */
    47.             Random random = new Random();
    48.             try {
    49.                 Thread.sleep(random.nextInt(100));
    50.             } catch (InterruptedException e) {
    51.                 ;
    52.             }
    53.         }
    54.     };
    55.  
    56.     public void init() {
    57.         indexingThread.start();
    58.     }
    59.  
    60.     public void shutdown() {
    61.         indexingThread.terminate();
    62.     }
    63.  
    64.     public void saveAttachment(InputStream in, String documentId,
    65.                                String originalFileName) throws Exception {
    66.         File file = saveAsFile(in, documentId, originalFileName);
    67.         try {
    68.             channel.put(file);
    69.         } catch (InterruptedException e) {
    70.             ;
    71.         }
    72.         indexingThread.terminationToken.reservations.incrementAndGet();
    73.     }
    74.  
    75.     private File saveAsFile(InputStream in, String documentId,
    76.                             String originalFileName) throws IOException {
    77.         String dirName = ATTACHMENT_STORE_BASE_DIR + documentId;
    78.         File dir = new File(dirName);
    79.         dir.mkdirs();
    80.         File file = new File(dirName + "/"
    81.                 + Normalizer.normalize(originalFileName, Normalizer.Form.NFC));
    82.  
    83.         // 防止目录跨越攻击
    84.         if (!dirName.equals(file.getCanonicalFile().getParent())) {
    85.             throw new SecurityException("Invalid originalFileName:" + originalFileName);
    86.         }
    87.  
    88.         BufferedOutputStream bos = null;
    89.         BufferedInputStream bis = new BufferedInputStream(in);
    90.         byte[] buf = new byte[2048];
    91.         int len = -1;
    92.         try {
    93.             bos = new BufferedOutputStream(new FileOutputStream(file));
    94.             while ((len = bis.read(buf)) > 0) {
    95.                 bos.write(buf, 0, len);
    96.             }
    97.             bos.flush();
    98.         } finally {
    99.             try {
    100.                 bis.close();
    101.             } catch (IOException e) {
    102.                 ;
    103.             }
    104.             try {
    105.                 if (null != bos) {
    106.                     bos.close();
    107.                 }
    108.             } catch (IOException e) {
    109.                 ;
    110.             }
    111.         }
    112.         return file;
    113.     }
    114. }

     

    通道积压

    消费者处理过慢时会出现通道积压，需要进行处理，分以下两种：

    1.使用有界阻塞队列：

    ArrayBlockingQueue和有容量限制的LinkedBlockingQueue

    2.使用带流量控制的无界阻塞队列:

    不带容量控制的LinkedBlockingQueue。借助流量控制实现，对同一时间内可有多少个生产者线程往通道中存储产品进行限制。本例使用基于Semaphore的支持流量控制的实现

    1. package comsumerproducer;
    2.  
    3. import java.util.concurrent.BlockingQueue;
    4. import java.util.concurrent.Semaphore;
    5.  
    6. /**
    7.  * 基于Semaphore的支持流量控制的通道实现
    8.  * @param 
    9.  */
    10. public class SemaphoreBasedChannel
       implements Channel
       {
    11.     private final BlockingQueue
       queue;
    12.     private final Semaphore semaphore;
    13.  
    14.     public SemaphoreBasedChannel(BlockingQueue
       queue, int flowLimit) {
    15.         this.queue = queue;
    16.         this.semaphore = new Semaphore(flowLimit);
    17.     }
    18.  
    19.     @Override
    20.     public P take() throws InterruptedException {
    21.         return queue.take();
    22.     }
    23.  
    24.     @Override
    25.     public void put(P product) throws InterruptedException {
    26.         semaphore.acquire();
    27.         try {
    28.             queue.put(product);
    29.         } finally {
    30.             semaphore.release();
    31.         }
    32.     }
    33. }

    优化的工作窃取算法思想

    producer-consumer中通常channel用queue实现，一个通道可对应一个或多个队列实例。现在本例仅使用一个ArrayBlockingQueue，如果有多个消费者线程从这个queue中获取产品，共享同一个实例。导致锁的竞争。

    如果一个通道实例对应多个队列实例，就可以实现多个消费者线程从通道中取产品时候访问各自的队列实例。

    如果一个消费者从自己的队列中取完任务，可以继续从其他消费者的队列中取出产品进行处理。

    1. package comsumerproducer;
    2.  
    3. import java.util.concurrent.BlockingDeque;
    4.  
    5. public interface WorkStealingEnableChannel
       extends Channel
       {
    6.     P take(BlockingDeque
       preferredQueue) throws InterruptedException;
    7. }
    8.  
    1. package comsumerproducer;
    2.  
    3. import java.util.concurrent.BlockingDeque;
    4.  
    5. public class WorkStealingChannel implements WorkStealingEnableChannel {
    6.     /**
    7.      * 受管队列
    8.      * @param preferredQueue
    9.      * @return
    10.      * @throws InterruptedException
    11.      */
    12.     private final BlockingDeque[] managedQueue;
    13.  
    14.     public WorkStealingChannel(BlockingDeque[] managedQueue) {
    15.         this.managedQueue = managedQueue;
    16.     }
    17.  
    18.     @Override
    19.     public T take(BlockingDeque preferredQueue) throws InterruptedException {
    20.         /**
    21.          * 优先从指定的受管队列中取产品
    22.          */
    23.         BlockingDeque targetQueue = preferredQueue;
    24.         T product = null;
    25.  
    26.         /**
    27.          * 试图从指定的队列队首取"产品"
    28.          */
    29.         if (null != targetQueue) {
    30.             product = targetQueue.poll();
    31.         }
    32.  
    33.         int queueIndex = -1;
    34.         while (null == product) {
    35.             queueIndex = (queueIndex + 1) % managedQueue.length;
    36.             targetQueue = managedQueue[queueIndex];
    37.             /**
    38.              * 试图从其他受管队列的队尾取
    39.              */
    40.             product = targetQueue.pollLast();
    41.             if (preferredQueue == targetQueue) {
    42.                 break;
    43.             }
    44.         }
    45.  
    46.         if (null == product) {
    47.             /**
    48.              * 随机窃取其他受管队列的产品
    49.              */
    50.             queueIndex = (int) (System.currentTimeMillis() % managedQueue.length);
    51.             targetQueue = managedQueue[queueIndex];
    52.             product = targetQueue.pollLast();
    53.             System.out.println("stealed from " + queueIndex + ":" + product);
    54.         }
    55.         return product;
    56.     }
    57.  
    58.     @Override
    59.     public T take() throws InterruptedException {
    60.         return take(null);
    61.     }
    62.  
    63.     @Override
    64.     public void put(T product) throws InterruptedException {
    65.         int targetIndex = (product.hashCode() % managedQueue.length);
    66.         BlockingDeque targetQueue = managedQueue[targetIndex];
    67.         targetQueue.put(product);
    68.     }
    69. }
    1. package comsumerproducer;
    2.  
    3. import twophased.AbstractTerminatedThread;
    4. import twophased.TerminationToken;
    5.  
    6. import java.util.Random;
    7. import java.util.concurrent.BlockingDeque;
    8. import java.util.concurrent.LinkedBlockingDeque;
    9.  
    10. /**
    11.  * 工作窃取算法
    12.  */
    13. public class WorkStealingExample {
    14.     private final WorkStealingEnableChannel channel;
    15.  
    16.     private final TerminationToken token = new TerminationToken();
    17.  
    18.     public WorkStealingExample() {
    19.         int nCPU = Runtime.getRuntime().availableProcessors();
    20.         int consumerCount = nCPU / 2 + 1;
    21.  
    22.         BlockingDeque[] managedQueue = new LinkedBlockingDeque[consumerCount];
    23.  
    24.         /**
    25.          * 该通道实例对应了多个queue
    26.          */
    27.         channel = new WorkStealingChannel(managedQueue);
    28.  
    29.         Consumer[] consumers = new Consumer[consumerCount];
    30.  
    31.         for (int i = 0; i < consumerCount; i++) {
    32.             managedQueue[i] = new LinkedBlockingDeque();
    33.             consumers[i] = new Consumer(token, managedQueue[i]);
    34.         }
    35.  
    36.         for (int i = 0; i < nCPU; i++) {
    37.             new Producer().start();
    38.         }
    39.  
    40.         for (int i = 0; i < consumerCount; i++) {
    41.             consumers[i].start();
    42.         }
    43.     }
    44.  
    45.     public void doSomething() {
    46.  
    47.     }
    48.  
    49.     public static void main(String[] args) throws InterruptedException {
    50.         WorkStealingExample wse = new WorkStealingExample();
    51.  
    52.         wse.doSomething();
    53.         Thread.sleep(3500);
    54.     }
    55.  
    56.     private class Producer extends AbstractTerminatedThread {
    57.         private int i = 0;
    58.  
    59.         @Override
    60.         protected void doRun() throws Exception {
    61.             channel.put(String.valueOf(i++));
    62.             token.reservations.incrementAndGet();
    63.         }
    64.     }
    65.  
    66.     private class Consumer extends AbstractTerminatedThread {
    67.         private final BlockingDeque workQueue;
    68.  
    69.         public Consumer(TerminationToken token, BlockingDeque workQueue) {
    70.             super(token);
    71.             this.workQueue = workQueue;
    72.         }
    73.  
    74.         @Override
    75.         protected void doRun() throws Exception {
    76.             /**
    77.              * WorkStealingEnableChannel接口的take方法实现了工作窃取算法
    78.              */
    79.             String produce = channel.take(workQueue);
    80.             System.out.println("processing product:" + produce);
    81.  
    82.             /**
    83.              * 模拟执行真正操作的时间消耗
    84.              */
    85.             try {
    86.                 Thread.sleep(new Random().nextInt(50));
    87.             } catch (InterruptedException e) {
    88.                 ;
    89.             } finally {
    90.                 token.reservations.decrementAndGet();
    91.             }
    92.         }
    93.     }
    94. }

  • 相关阅读:
    PHP环境安装
    类加载机制和类加载器
    Json、JDBC
    docker mysql 启动报错
    java虚拟机 JVM问题记录
    【AI赋能医学】基于深度学习和HRV特征的多类别心电图分类
    Linux下提取文件夹下的所有文件名称
    Mysql数据库基础
    CentOS 7 基于C 连接ZooKeeper 客户端
    美创科技入选IDC中国等保合规市场报告推荐厂商
  • 原文地址：https://blog.csdn.net/wanglt311/article/details/126695174