生產者消費者模式之工作竊取算法 1、一個通道只有一個隊列，多個消費者共享一個隊列實例，導致鎖的競爭，如果一個通道擁有對個隊列，則消費者可以從通道中獲取各自隊列獲取數據。 2、如要服務有高性能和可靠性的要求，Consumer-PRoducer模式請使用 kafka等開源工具
public interface WorkStealingEnableChannel<P> extends Chanel<P> { P take(BlockingDeque<P> preferredQueue) throws InterruptedException;}public class WorkStealingChannel<P> implements WorkStealingEnableChannel<P> { //雙端隊列，可以從2端插入值或獲取值，繼承了BlockingQueue private final BlockingDeque<P>[] managedQueues; public WorkStealingChannel(BlockingDeque<P>[] managedQueues) { super(); this.managedQueues = managedQueues; } @Override public P take() throws InterruptedException { return take(null); } @Override public void put(P product) throws InterruptedException { int targetIndex = (product.hashCode() % managedQueues.length); BlockingQueue<P> targetQueue = managedQueues[targetIndex]; targetQueue.put(product); } @Override public P take(BlockingDeque<P> preferredQueue) throws InterruptedException { BlockingDeque<P> targetQueue = preferredQueue; P product = null; //優先從指定的隊列獲取值 if(null != targetQueue){ product = targetQueue.poll(); } int queueIndex = -1; while(null != product){ queueIndex = (queueIndex +1) % managedQueues.length; targetQueue = managedQueues[queueIndex]; //試圖從其他受管隊列的隊尾“竊取”“產品” product = targetQueue.pollLast(); if(preferredQueue == targetQueue){ break; } } if(null == product){ //隨機竊取 其他受管隊列的產品 queueIndex = (int) (System.currentTimeMillis() % managedQueues.length); targetQueue = managedQueues[queueIndex]; product = targetQueue.pollLast(); System.out.println("stealed from " + queueIndex + ": " + product); } return product; }}public class WorkStealingExample { private final WorkStealingEnableChannel<String> channel; private final TerminationToken token = new TerminationToken(); public static void main(String[] args) throws InterruptedException { WorkStealingExample wse = new WorkStealingExample(); //Thread.sleep(3500); } public WorkStealingExample(){ int nCPU = Runtime.getRuntime().availableProcessors(); int consumerCount = nCPU/2 + 1; BlockingDeque<String>[] managedQueues = new LinkedBlockingDeque[consumerCount]; channel = new WorkStealingChannel<String>(managedQueues); Consumer[] consumers = new Consumer[consumerCount]; for(int i=0; i<consumerCount; i++){ managedQueues[i] = new LinkedBlockingDeque<String>(); consumers[i] = new Consumer(token, managedQueues[i]); } for(int i=0; i<nCPU; i++){ new Producer().start(); } for(int i=0; i<consumerCount; i++){ consumers[i].start(); } } private class Producer extends AbstractTerminatableThread{ private int i = 0; @Override protected void doRun() throws Exception { channel.put(String.valueOf(i++)); Thread.sleep(10); token.reservations.incrementAndGet(); } } private class Consumer extends AbstractTerminatableThread{ private final BlockingDeque<String> workQueue; public Consumer(TerminationToken token, BlockingDeque<String> workQueue) { super(token); this.workQueue = workQueue; } @Override protected void doRun() throws Exception { /** * 實現了工作竊取算法 */ String product = channel.take(); if(product != null){ } System.out.println("Processing product:" + product); try { Thread.sleep(new Random().nextInt(50)); } catch (Exception e) { }finally{ token.reservations.decrementAndGet(); } } }}新聞熱點
疑難解答
