java JUC包小结
?
package com.mike.juc;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;import java.util.concurrent.TimeUnit;/** * TODO Comment of CyclicBarrierTest */public class CyclicBarrierTest { final int N; final int[][] data; final CyclicBarrier barrier; private int[] result; class Worker implements Runnable { private int index; private int[] oneRow; Worker(int[] data, int index) { this.index = index; this.oneRow = data; } public void run() { try { TimeUnit.MILLISECONDS.sleep((long) (Math.random() * 10000)); // Thread.sleep((long) (Math.random() * 10000)); } catch (InterruptedException e) { } System.out.println("start " + index + " row!"); int rowResult = 0; for (int i = 0, length = oneRow.length; i < length; i++) { rowResult += oneRow[i]; } result[index] = rowResult; try { System.out.println("finish " + index + " row, result = " + rowResult); barrier.await(); } catch (InterruptedException ex) { return; } catch (BrokenBarrierException ex) { return; } } } public CyclicBarrierTest(int[][] matrix) { data = matrix; N = matrix.length; result = new int[N]; barrier = new CyclicBarrier(N, new Runnable() { public void run() { int total = 0; for (int i = 0, length = result.length; i < length; i++) { total += result[i]; } System.out.println("total result:" + total); } }); } public void start() { for (int i = 0; i < N; ++i) { Thread thread = new Thread(new Worker(data[i], i)); // thread.setDaemon(true); thread.start(); } System.out.println("abc"); } public static void main(String[] args) { int[][] matrix = { { 1, 2, 3 }, { 2, 3, 4 }, { 3, 4, 5 } }; new CyclicBarrierTest(matrix).start(); }}?
package com.mike.juc;import java.util.concurrent.Exchanger;/** * TODO Comment of ExchangerTest */public class ExchangerTest { Exchanger<String> exchanger = new Exchanger<String>(); class FillingLoop implements Runnable { public void run() { String currentBuffer = "123"; System.out.println("===" + Thread.currentThread().getId() + "\tcurrentBuffer=" + currentBuffer); try { currentBuffer = exchanger.exchange(currentBuffer); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("===" + Thread.currentThread().getId() + "\tcurrentBuffer=" + currentBuffer); } } class EmptyingLoop implements Runnable { public void run() { String currentBuffer = "124"; System.out.println(Thread.currentThread().getId() + "\tcurrentBuffer=" + currentBuffer); try { currentBuffer = exchanger.exchange(currentBuffer); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getId() + "\tcurrentBuffer=" + currentBuffer); } } void start() { new Thread(new FillingLoop()).start(); new Thread(new EmptyingLoop()).start(); } public static void main(String[] args) { new ExchangerTest().start(); }}?测试?LockSupport，这个类提供了最基本的wait/notify机制。可为各类wait/notify的实现提供基础
package com.mike.juc;import java.util.Queue;import java.util.concurrent.ConcurrentLinkedQueue;import java.util.concurrent.atomic.AtomicBoolean;import java.util.concurrent.locks.LockSupport;/** * TODO Comment of LockSupportTest * */public class LockSupportTest { private AtomicBoolean locked = new AtomicBoolean(false); private Queue<Thread> waiters = new ConcurrentLinkedQueue<Thread>(); public void lock() { boolean wasInterrupted = false; Thread current = Thread.currentThread(); waiters.add(current); // Block while not first in queue or cannot acquire lock while (waiters.peek() != current || !locked.compareAndSet(false, true)) { LockSupport.park(); if (Thread.interrupted()) // ignore interrupts while waiting wasInterrupted = true; } waiters.remove(); if (wasInterrupted) // reassert interrupt status on exit current.interrupt(); } public void unlock() { locked.set(false); LockSupport.unpark(waiters.peek()); } public static void main(String[] args) { LockSupportTest lockSupportTest = new LockSupportTest(); for (int i = 0; i < 2; i++) { new Thread(new Runlock(lockSupportTest), "i=" + i).start(); } lockSupportTest.unlock(); }}class Runlock implements Runnable { private LockSupportTest lockSupportTest; /** * @param lockSupportTest */ public Runlock(LockSupportTest lockSupportTest) { this.lockSupportTest = lockSupportTest; } @Override public void run() { lockSupportTest.lock(); }}?
package com.mike.juc;import java.util.concurrent.TimeUnit;/** * TODO Comment of ObjectTest */public class ObjectTest { private Object obj = new Object(); public void getAndWait() { System.out .println("thread id" + Thread.currentThread().getId() + "\tgetAndWait position 0"); synchronized (obj) { System.out.println("thread id" + Thread.currentThread().getId() + "\tgetAndWait position 1"); try { obj.wait(); } catch (InterruptedException e) { } System.out.println("thread id" + Thread.currentThread().getId() + "\tgetAndWait position 2"); } System.out .println("thread id" + Thread.currentThread().getId() + "\tgetAndWait position 4"); } public void notifyOneThread() { synchronized (obj) { try { obj.notify(); System.out.println("thread id" + Thread.currentThread().getId() + "\tnotifyOneThread"); } catch (Exception e) { } } } public static void main(String[] args) throws InterruptedException { System.err.println("main 1"); ObjectTest objectTest = new ObjectTest(); for (int i = 0; i < 10; i++) { new Thread(new WaitThread(objectTest)).start(); } System.err.println("main 2"); TimeUnit.SECONDS.sleep(10l); System.err.println("main 3"); for (int i = 0; i < 10; i++) { objectTest.notifyOneThread(); TimeUnit.SECONDS.sleep(1l); System.err.println("main 3.5"); } System.err.println("main 4"); }}class WaitThread implements Runnable { private ObjectTest objectTest; public WaitThread(ObjectTest objectTest) { this.objectTest = objectTest; } @Override public void run() { objectTest.getAndWait(); }}//从运行结果看，notify的解锁是有序的，先到先解//main 1//thread id8 getAndWait position 0//thread id8 getAndWait position 1//thread id9 getAndWait position 0//thread id9 getAndWait position 1//thread id10 getAndWait position 0//thread id10 getAndWait position 1//thread id11 getAndWait position 0//thread id11 getAndWait position 1//thread id12 getAndWait position 0//thread id12 getAndWait position 1//thread id13 getAndWait position 0//thread id13 getAndWait position 1//thread id14 getAndWait position 0//thread id14 getAndWait position 1//thread id15 getAndWait position 0//thread id15 getAndWait position 1//thread id16 getAndWait position 0//thread id16 getAndWait position 1//main 2//thread id17 getAndWait position 0//thread id17 getAndWait position 1//main 3//thread id1 notifyOneThread//thread id8 getAndWait position 2//thread id8 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id9 getAndWait position 2//thread id9 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id10 getAndWait position 2//thread id10 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id11 getAndWait position 2//thread id11 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id12 getAndWait position 2//thread id12 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id13 getAndWait position 2//thread id13 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id14 getAndWait position 2//thread id14 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id15 getAndWait position 2//thread id15 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id16 getAndWait position 2//thread id16 getAndWait position 4//main 3.5//thread id1 notifyOneThread//thread id17 getAndWait position 2//thread id17 getAndWait position 4//main 3.5//main 4
package com.mike.juc;import java.util.concurrent.Semaphore;/** * TODO Comment of Pool */public class Pool { private static final int MAX_AVAILABLE = 100; private final Semaphore available = new Semaphore(MAX_AVAILABLE, true); public Object getItem() throws InterruptedException { available.acquire(); return getNextAvailableItem(); } public void putItem(Object x) { if (markAsUnused(x)) available.release(); } // Not a particularly efficient data structure; just for demo protected Object[] items = new String[MAX_AVAILABLE]; protected boolean[] used = new boolean[MAX_AVAILABLE]; protected synchronized Object getNextAvailableItem() { for (int i = 0; i < MAX_AVAILABLE; ++i) { if (!used[i]) { used[i] = true; return items[i]; } } return null; // not reached } protected synchronized boolean markAsUnused(Object item) { for (int i = 0; i < MAX_AVAILABLE; ++i) { if (item == items[i]) { if (used[i]) { used[i] = false; return true; } else return false; } } return false; }}
