zoukankan      html  css  js  c++  java

  • 【并发编程】使用BlockingQueue实现<多生产者，多消费者>




     MasterThread：
    1.  持有一个BlockingQueue队列，用于并发接收存储MetaData对象;
    2.  使用Hash一致性算法ketama，来选择SlaveThread节点;
    3.  从BlockingQueue队列中，取出MetaData对象，分配给各SlaveThread节点；
    4.  SlaveThread节点负责真正处理MetaData对象；
         
     SlaveThread：
    1. 持有一个BlockingQueue队列，用于存储MetaData对象;
    2. 负责真正处理MetaData对象;

    Ketama：
     一致性Hash算法，是一种分布式算法，常用于负载均衡；

    测试说明：
    1. 假设处理每个MetaData对象需要耗时0.5秒，需要处理500个MetaData对象；
    2. 若是串行处理，则需要0.5*500=250秒；
    3. 使用上图所示的master/slave算法，则可以5秒内完成；


    MasterThread.java

    核心程序介绍

    SlaveThread.java

    核心程序

    测试

    执行结果：
    省略......
    省略......

    完整程序

    HashAlgorithm.java
    1. package ll.concurrent.BlockingQueue.Ketama;
    3. import java.io.UnsupportedEncodingException;
    4. import java.security.MessageDigest;
    5. import java.security.NoSuchAlgorithmException;
    7. public enum HashAlgorithm {
    8.   KETAMA_HASH;
    9.   private static final String UTF_8 = "UTF-8";
    10.   private static final String MD5 = "MD5";
    12.   public long hash(byte[] digest, int nTime) {
    13.     long rv = ((long) (digest[3 + nTime * 4] & 0xFF) << 24)
    14.         | ((long) (digest[2 + nTime * 4] & 0xFF) << 16)
    15.         | ((long) (digest[1 + nTime * 4] & 0xFF) << 8)
    16.         | (digest[0 + nTime * 4] & 0xFF);
    17.     return rv & 0xffffffffL; /* Truncate to 32-bits */
    18.   }
    20.   public byte[] computeMd5(String key) {
    21.     MessageDigest md5;
    22.     try {
    23.       md5 = MessageDigest.getInstance(MD5);
    24.     } catch (NoSuchAlgorithmException e) {
    25.       throw new RuntimeException("MD5 not supported", e);
    26.     }
    27.     md5.reset();
    28.     byte[] keyBytes = null;
    29.     try {
    30.       keyBytes = key.getBytes(UTF_8);
    31.     } catch (UnsupportedEncodingException e) {
    32.       throw new RuntimeException("Unknown string :" + key, e);
    33.     }
    34.     md5.update(keyBytes);
    35.     return md5.digest();
    36.   }
    37. }
    KetamaNodeLocator.java
    1. package ll.concurrent.BlockingQueue.Ketama;
    3. import java.util.Collection;
    4. import java.util.TreeMap;
    6. /**
    7.  * 一致性Hash算法：是一种分布式算法，常用于负载均衡;
    8.  *
    9.  * @param <T>
    10.  */
    11. public final class KetamaNodeLocator<T> {
    12.   private TreeMap<Long, T> ketamaNodes;
    13.   private HashAlgorithm hashAlg;
    14.   private int numReps = 160;
    16.   public KetamaNodeLocator(Collection<T> nodes) {
    17.     this(nodes, HashAlgorithm.KETAMA_HASH);
    18.   }
    20.   public KetamaNodeLocator(Collection<T> nodes, HashAlgorithm alg) {
    21.     this(nodes, HashAlgorithm.KETAMA_HASH, 160);
    22.   }
    24.   public KetamaNodeLocator(Collection<T> nodes, HashAlgorithm alg,
    25.       int nodeCopies) {
    26.     hashAlg = alg;
    27.     ketamaNodes = new TreeMap<Long, T>();
    28.     numReps = nodeCopies;
    29.     for (T node : nodes) {
    30.       for (int i = 0; i < numReps / 4; i++) {
    31.         byte[] digest = hashAlg.computeMd5(node.toString() + i);
    32.         for (int h = 0; h < 4; h++) {
    33.           long m = hashAlg.hash(digest, h);
    34.           ketamaNodes.put(m, node);
    35.         }
    36.       }
    37.     }
    38.   }
    40.   public T getNode(final String k) {
    41.     byte[] digest = hashAlg.computeMd5(k);
    42.     T rv = getNodeForKey(hashAlg.hash(digest, 0));
    43.     return rv;
    44.   }
    46.   T getNodeForKey(long hash) {
    47.     if (ketamaNodes.isEmpty()) {
    48.       return null;
    49.     }
    50.     if (!ketamaNodes.containsKey(hash)) {
    51.       Object ceilValue = ((TreeMap<Long, T>) ketamaNodes)
    52.           .ceilingKey(hash);
    53.       if (ceilValue != null) {
    54.         try {
    55.           hash = Long.valueOf(ceilValue.toString());
    56.         } catch (NumberFormatException e) {
    57.           e.printStackTrace();
    58.           hash = 0;
    59.         }
    60.       }
    61.       if (ceilValue == null || hash == 0) {
    62.         hash = ketamaNodes.firstKey();
    63.       }
    64.     }
    65.     return ketamaNodes.get(hash);
    66.   }
    68. }
    MasterThread.java
    1. package ll.concurrent.BlockingQueue;
    3. import java.util.HashMap;
    4. import java.util.Map;
    5. import java.util.concurrent.BlockingQueue;
    6. import java.util.concurrent.ExecutorService;
    7. import java.util.concurrent.Executors;
    8. import java.util.concurrent.LinkedBlockingQueue;
    10. import ll.concurrent.BlockingQueue.Ketama.KetamaNodeLocator;
    12. /**
    13.  * <pre>
    14.  * MasterThread：
    15.  *  1. 持有一个BlockingQueue队列,用于并发接收存储MetaData对象;
    16.  *  2. 使用Hash一致性算法ketama来选择SlaveThread节点;
    17.  *  3. 从BlockingQueue队列中，取出MetaData对象，分配给SlaveThread节点；
    18.  *  4. SlaveThread节点负责真正处理MetaData对象；
    19.  *     
    20.  * SlaveThread：
    21.  *  1. 持有一个BlockingQueue队列,用于存储MetaData对象;
    22.  *  2. 负责真正处理MetaData对象;
    23.  * </pre>
    24.  */
    25. public class MasterThread {
    26.   private static int SLAVE_ENGINE_NUMBER_MAX = 100;
    27.   private static int _BLOCKSIZE = 5000;
    28.   private static MasterThread masterThread;
    30.   private BlockingQueue<MetaData> metaDataQueue;
    31.   private SlaveEngineThread slaveEngineThread;
    33.   public static synchronized MasterThread getInstance() {
    34.     if (masterThread == null) {
    35.       masterThread = new MasterThread();
    36.     }
    37.     return masterThread;
    38.   }
    40.   private MasterThread() {
    41.     metaDataQueue = new LinkedBlockingQueue<MetaData>(_BLOCKSIZE);
    42.     startSlaveThreadEngine();
    43.   }
    44.   
    45.   private void startSlaveThreadEngine() {
    46.     slaveEngineThread = new SlaveEngineThread(SLAVE_ENGINE_NUMBER_MAX);
    47.     slaveEngineThread.start();
    48.   }
    50.   public synchronized void put(MetaData object) {
    51.     if (object == null)
    52.       return;
    54.     if (!metaDataQueue.offer(object)) {
    55.       System.err.println("BlockingQueue is up to max size:"
    56.           + metaDataQueue.size());
    57.     }
    58.   }
    60.   private class SlaveEngineThread extends Thread {
    61.     private ExecutorService executorService;
    62.     private Map<Integer, SlaveThread> slaveThreadMap;
    64.     //本示例采用一致性Hash算法，选择SlaveThread
    65.     private KetamaNodeLocator<Integer> nodeLocator;
    67.     public SlaveEngineThread(final int nThreads) {
    68.       slaveThreadMap = new HashMap<Integer, SlaveThread>();
    70.       // 创建线程池，并发执行SlaveThread
    71.       executorService = Executors.newFixedThreadPool(nThreads);
    72.       for (int i = 0; i < nThreads; i++) {
    73.         SlaveThread command = new SlaveThread(i);
    74.         executorService.execute(command);
    75.         slaveThreadMap.put(i, command);
    76.       }
    77.       nodeLocator = new KetamaNodeLocator<Integer>(
    78.           slaveThreadMap.keySet());
    79.     }
    81.     @Override
    82.     public void run() {
    83.       while (true) {
    84.         MetaData metaData = null;
    85.         try {
    86.           // 堵塞获取
    87.           metaData = metaDataQueue.take();
    89.           // 通过hash算法获取slaveThread编号
    90.           Integer nodeNumber = nodeLocator.getNode(metaData.getId());
    91.           SlaveThread slaveThread = slaveThreadMap.get(nodeNumber);
    92.           if (slaveThread != null) {
    93.             // 将MetaData存入SlaveThread处理
    94.             slaveThread.put(metaData);
    95.           }
    96.         } catch (InterruptedException e) {
    97.           executorService.shutdown();
    98.           System.err.println("SlaveEngineThread is Interrupted ... ");
    99.           break;
    100.         } catch (Exception e) {
    101.           System.err.println("failed to handle meta data" + e);
    102.         }
    103.       }
    104.       System.err.println("Master Thread exit...");
    105.     }
    106.   }
    108.   public void exit() {
    109.     stopSlaveEngineThread();
    110.   }
    112.   private void stopSlaveEngineThread() {
    113.     slaveEngineThread.interrupt();
    114.   }
    115. }
    SlaveThread.java
    1. package ll.concurrent.BlockingQueue;
    3. import java.text.SimpleDateFormat;
    4. import java.util.concurrent.BlockingQueue;
    5. import java.util.concurrent.LinkedBlockingQueue;
    7. public class SlaveThread implements Runnable {
    8.   private BlockingQueue<MetaData> msgQueue = new LinkedBlockingQueue<MetaData>(
    9.       200);
    11.   private int slaveThreadID;
    13.   public SlaveThread() {
    14.   }
    16.   public SlaveThread(int slaveThreadID) {
    17.     super();
    18.     this.slaveThreadID = slaveThreadID;
    19.   }
    21.   public void run() {
    22.     while (true) {
    23.       MetaData metaData = null;
    24.       try {
    25.         // 堵塞获取
    26.         metaData = msgQueue.take();
    27.         handleMetaData(metaData);
    28.       } catch (InterruptedException e) {
    29.         System.err.println("SlaveThread[" + this.slaveThreadID + "] is Interrupted...");
    30.         break;
    31.       } catch (Exception e) {
    32.         System.err.println("failed to handle meta data" + e);
    33.       }
    34.     }
    35.     System.err.println("SlaveThread[" + this.slaveThreadID + "] exit...");
    36.   }
    38.   public void put(MetaData object) {
    39.     if (object == null)
    40.       return;
    41.     if (!msgQueue.offer(object)) {
    42.       System.err.println("SlaveThread BlockingQueue up to max size");
    43.     }
    44.   }
    46.   private void handleMetaData(MetaData metaData) throws Exception {
    47.     // 模拟处理，耗时500毫秒
    48.     Thread.sleep(500);
    49.     System.out.println("SlaveThread["+ this.slaveThreadID + "] "
    50.         + metaData.toString()
    51.         + new SimpleDateFormat("HH:mm:ss").format(System
    52.             .currentTimeMillis()));
    53.   }
    55. }
    MetaData.java
    1. package ll.concurrent.BlockingQueue;
    3. public class MetaData {
    4.   private String id;
    5.   private String desc;
    7.   public MetaData() {
    8.     super();
    9.   }
    11.   public MetaData(String id) {
    12.     super();
    13.     this.id = id;
    14.   }
    15.   
    16.   public MetaData(String id, String desc) {
    17.     super();
    18.     this.id = id;
    19.     this.desc = desc;
    20.   }
    22.   public String getId() {
    23.     return id;
    24.   }
    26.   public void setId(String id) {
    27.     this.id = id;
    28.   }
    30.   public String getDesc() {
    31.     return desc;
    32.   }
    34.   public void setDesc(String desc) {
    35.     this.desc = desc;
    36.   }
    38.   @Override
    39.   public String toString() {
    40.     return "MetaData [id=" + id + ", desc=" + desc + "]";
    41.   }
    42. }
    TestCase.java
    1. package ll.concurrent.BlockingQueue;
    3. import java.text.SimpleDateFormat;
    5. public class TestCase {
    6.   public static void main(String[] args) throws InterruptedException {
    7.     MasterThread masterThread = MasterThread.getInstance();
    9.     System.out.println("Start time:"
    10.         + new SimpleDateFormat("HH:mm:ss").format(System
    11.             .currentTimeMillis()));
    12.     /**
    13.      * 每个MetaData都需要0.5S的处理时间，如果串行执行，则需要500*0.5=250; 
    14.      * 现采用并行处理，只需要很短的时间即可执行完；
    15.      */
    16.     for (int i = 0; i < 500; i++) {
    17.       MetaData metaData = new MetaData(i + "");
    18.       masterThread.put(metaData);
    19.     }
    20.   }
    22. }

    参考：
    《Java编发编程实战》：7.2.1 日志服务章节： <N生产者，1个消费者> 
  • 相关阅读:
    Pandas Statistical Functions
    pyspark 记录
    走出浮躁的泥沼：学会享受学习过程的乐趣
    寻找知识不变的那部分
    走出浮躁的泥沼：把一件事做到某种境界
    走出浮躁的泥沼：为什么会浮躁
    对抗拖延症最直接有效的方法
    实施“番茄工作法”需要注意的一些细节
    东西学了容易忘？学会跟踪你的知识
    再谈读书与信息的获取、沉淀
  • 原文地址：https://www.cnblogs.com/ssslinppp/p/6279796.html
Copyright © 2011-2022 走看看