并发编程之基础( 四)

新类库

前面已经把并发编程的基础知识讲的差不多了，这章主要介绍一下JAVA中其它一些关于并发编程的类库,主要有一下几个类库。

• CountDownLatch
• CyclicBarrier
• BlockingQueue
• ScheduleExecutor
• Semaphore
• Exchanger

---

1. CountDownLatch

　　该类主要是同步一个或多个任务，强制一个或多个任务等待其它任务执行的一组操作完成。可以给该对象设置一个初始计数值，当计数值不为0时，调用该对象的await()方法就会阻塞，调用counDown()方法会让计数值减1，当计数值为0时阻塞任务会被唤醒。其典型用法就是将一个程序分成多个独立的任务，并给CountDownLatch设定一个初始值，该初始值应该为首先需要执行的线程的个数(比如赛跑，5个运动员都做好准备之后，裁判才能打枪，这时初始值应该设置为5)。一些任务需要等待其它任务先完成或者其它任务的一部分完成，那么可以待用await()将自己挂起。而另一些任务的某些操作完成时调用countDown()方法来减小计数值，等待计数值为0时，挂起的任务则则认为当前所有的条件以满足继续执行的需要了，则可以继续运行。注意：计数值只能被设置一次且在new的时候就要指定初值，而且该对象只能使用一次，如果想重复使用，请考虑CyclicBarrier

package com.dy.xidian;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class TaskPortion implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private static Random rand = new Random(47);
    private final CountDownLatch latch;

    public TaskPortion(CountDownLatch latch) {
        super();
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            doWork();
            latch.countDown();
        } catch (InterruptedException e) {
        }

    }

    public void doWork() throws InterruptedException {
        TimeUnit.MILLISECONDS.sleep(rand.nextInt(2000));
        System.out.println(this + "completed");
    }

    public String toString() {
        return String.format("%1$-3d", id);
    }
}

class WaitingTask implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private final CountDownLatch latch;

    public WaitingTask(CountDownLatch latch) {
        super();
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            latch.await();
            System.out.println("Latch barrier passed for " + this);
        } catch (InterruptedException e) {
            System.out.println(this + " interrupted");
        }
    }

    public String toString() {
        return String.format("WaitingTask %1$-3d ", id);
    }
}
public class CountDownLatchDemo {
    static final int SIZE = 100;
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(SIZE);
        for (int i = 0; i < 10; i++)
            exec.execute(new WaitingTask(latch));
        for (int i = 0; i < SIZE; i++)
            exec.execute(new TaskPortion(latch));
        System.out.println("Launched all tasks");
        exec.shutdownNow();
    }
}

2. CyclicBarrier

　　CyclicBarrier与CountDownLatch功能差不多，不同之处就是可以多次使用，等到计数值变为0时，它会自动重置。而且不需要每个线程都去调用类似countDown()这样的方法，因为每调用一个await()，它就会自动将计数值减1。它使用于这种情况：多个线程并行执行工作，大家一致向前推进，所有线程在这个阶段的工作都完成了(所有的线程都调用了await方法)，才能进入下一阶段，而对于那些早完成的线程只能先等待了。下面是一个赛马比赛，每个马可以看作一个线程，等所有的马都达到栅栏后，才能开始新一轮的比赛。

package com.dy.xidian;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class Horse implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private int strides = 0;
    private static Random rand = new Random(47);
    private static CyclicBarrier barrier;

    public Horse(CyclicBarrier b) {
        barrier = b;
    }

    public synchronized int getStriders() {
        return strides;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                synchronized (this) {
                    strides += rand.nextInt(3);
                }
                barrier.await();
            }
        } catch (InterruptedException e) {
            // TODO
        } catch (BrokenBarrierException e) {
            throw new RuntimeException(e);
        }
    }

    public String toString() {
        return "Horse " + id + " ";
    }

    public String tracks() {
        StringBuilder s = new StringBuilder();
        for (int i = 0; i < getStriders(); i++)
            s.append("*");
        s.append(id);
        return s.toString();
    }
}

public class HorseRace {
    static final int FINISH_LINE = 75;
    private List<Horse> horses = new ArrayList<Horse>();
    private ExecutorService exec = Executors.newCachedThreadPool();
    private CyclicBarrier barrier;

    public HorseRace(int nHorses, final int pause) {
        barrier = new CyclicBarrier(nHorses, new Runnable() {

            @Override
            public void run() {
                StringBuilder s = new StringBuilder();
                for (int i = 0; i < FINISH_LINE; i++) {
                    s.append("=");
                    System.out.println(s);
                    for (Horse horse : horses)
                        System.out.println(horse.tracks());
                    for (Horse horse : horses)
                        if (horse.getStriders() >= FINISH_LINE) {
                            System.out.println(horse + "won!");
                            exec.shutdownNow();
                            return;
                        }
                    try {
                        TimeUnit.MILLISECONDS.sleep(pause);
                    } catch (InterruptedException e) {
                        System.out.println("barrier-action sleep interrupted");
                    }
                }
            }
        });

        for (int i = 0; i < nHorses; i++) {
            Horse horse = new Horse(barrier);
            horses.add(horse);
            exec.execute(horse);
        }
    }

    public static void main(String[] args) {
        int nHorses = 7;
        int pause = 200;
        if (args.length > 0) {
            int n = new Integer(args[0]);
            nHorses = n > 0 ? n : nHorses;
        }
        if (args.length > 1) {
            int p = new Integer(args[1]);
            pause = p > -1 ? p : pause;
        }
        new HorseRace(nHorses, pause);
    }
}

运行结果：

=
**0
***1
*2
**3
*4
***5
***6
==
**0
***1
*2
**3
*4
***5
***6
===
**0
***1
*2
**3
*4
***5
***6
====

　　运行结果中的==表示栅栏，数字为每个马的编号，*的个数代表每个马目前跑了多少步。在代码我，我们可以看到在创建CyclicBarrier对象时，我们还给他传递了一个复写了Runnable后的对象，这是我CounDownLatch不同的地方。每当计数器的值为0的是时候，里面的该对象中的run方法会被调用。可能有这样一种情况，当计数值再次变为0时，上次的run方法还没执行完，它会不会创建新的线程重新执行run方法呢？通过测试，这种情况是不会发生的，只有等run执行完，才会去创建新的线程。

 3 BlockingQueue

BlockingQueue是一个接口，用于生产者-消费者模型，是一个线程安全的容器。它的实现类有LinkedBlockingQueue(空间无限,FIFO), ArrayBlockingQueue(空间有限,FIFO),PriorityBlockingQueue(元素等级高的在队头)，SynchronousQueue(内部没有缓冲区，生产者线程需要将产品直接交给一个空闲的消费者线程，否则将一直处于阻塞状态)

3.1 DelayQueue

　　DelayQueue是一个无界的阻塞队列，用于存放实现了Delayed接口的对象，其中的对象只能在其延迟期满才能从队列中取走。该队列的头部是延迟期满后保存时间最长的Delayed元素。如果没有任何延迟期满的对象，那就不会有任何头元素，这时如果使用take()方法从队列获取对象时会发生阻塞，使用poll时会直接返回null。

package com.dy.xidian;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class DelayedTask implements Runnable, Delayed {
    private static int counter = 0;
    private final int id = counter++;
    private final int delayTime;
    private final long trigger;

    protected static List<DelayedTask> sequeue = new ArrayList<DelayedTask>();

    // System.nanoTime()获取当前时间,结果是纳秒级
    // TimeUnit.MILLSECONDS.convert(time, TimeUnit.SECONDS)
    // 时间转换(一般是大单位转小单位),比如计算1s=多少ms之类的
    // time是时间,TimeUnit.SECONDS是原始单位(s),MILLISECONDS是转换后的单位(ms)
    public DelayedTask(int delayInMilliseconds) {
        delayTime = delayInMilliseconds;
        trigger = System.nanoTime()
                + TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS);
        sequeue.add(this);
    }

    // 重载Delayed接口的getDelay方法,该示例代码给出的是重载的标准形式
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(trigger - System.nanoTime(), TimeUnit.NANOSECONDS);

    }

    //比较每个对象的触发时间，以确定在队列中的位置
    @Override
    public int compareTo(Delayed arg) {
        DelayedTask that = (DelayedTask) arg;
        if (trigger < that.trigger)
            return -1;
        if (trigger > that.trigger)
            return 1;
        return 0;
    }

    @Override
    public void run() {
        System.out.println(this + " ");
    }

    public String toString() {
        return String.format("[%1$-4d]", counter) + "Task " + id;
    }

    public String summary() {
        return "(" + id + ":" + counter + ")";
    }

    public static class EndSentinel extends DelayedTask {
        private ExecutorService exec;

        public EndSentinel(int delay, ExecutorService e) {
            super(delay);
            exec = e;
        }

        public void run() {
            for (DelayedTask pt : sequeue) {
                System.out.println(pt.summary() + " ");
            }
            System.out.println(this + " Calling shutdownNow()");
            exec.shutdownNow();
        }
    }
}

class DelayTaskConsumer implements Runnable {
    private DelayQueue<DelayedTask> q;
    public DelayTaskConsumer(DelayQueue<DelayedTask> q) {
        this.q = q;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted())
                q.take().run();
        } catch (InterruptedException e) {
        }
        System.out.println("Finised DelayedTaskConsumer!");
    }

}
public class DelayQueueDemo {
    public static void main(String[] args) {
        Random rand = new Random(47);
        ExecutorService exec = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();
        for (int i = 0; i < 20; i++)
            queue.put(new DelayedTask(rand.nextInt(5000)));
        queue.add(new DelayedTask.EndSentinel(5000, exec));
        exec.execute(new DelayTaskConsumer(queue));
    }
}

运行结果：

[128 ]Task 11 
[200 ]Task 7 
[429 ]Task 5 
[520 ]Task 18 
[555 ]Task 1 
[961 ]Task 4 
[998 ]Task 16 
[1207]Task 9 
[1693]Task 2 
[1809]Task 14 
[1861]Task 3 
[2278]Task 15 
[3288]Task 10 
[3551]Task 12 
[4258]Task 0 
[4258]Task 19 
[4522]Task 8 
[4589]Task 13 
[4861]Task 17 
[4868]Task 6 
(0:21) 
(1:21) 
(2:21) 
(3:21) 
(4:21) 
(5:21) 
(6:21) 
(7:21) 
(8:21) 
(9:21) 
(10:21) 
(11:21) 
(12:21) 
(13:21) 
(14:21) 
(15:21) 
(16:21) 
(17:21) 
(18:21) 
(19:21) 
(20:21) 
[5000]Task 20 Calling shutdownNow()
Finised DelayedTaskConsumer!

　　该程序创建了20个delayedTask对象,这20对象其实是线程对象，然后将这20对象放入DelayedQueue中,同时将这20个对象加入到list中以表明创建的先后顺序。每个线程的延迟期是通过随机数指定的。在DelayedTask中有一个内部类，该类的作用就是遍历list，输出每个线程的信息(id + 延迟期)，最后关闭整个线程。DelayedTaskConsumer就是不断从DelayedQueue中取线程对象，然后让其执行。

　　关于Delayed接口的实现这里要强调一下，代码中写的是标准形式，也是策略模式的一种简单实现。delayTime是延迟期,需要我们指定。trigger表示这个对象的激活时间(比如到11点整时,其延迟期满)，其计算方法就是获取当前时间+延迟期。而getDelay(TimeUnit unit)这个函数是个关键，这个函数会被调用两次：第一次查看延期满的时间点和当前时间之差(比如当前时间9点，延迟期满是在11点),发现是正值，对象需要继续等待；第二次查看时发现是负值(比如当前时间已经到了12点了)，返回值为负数，说明对象的延迟期已经到了，可以使用了。 

3.2 PriorityBlockingQueue

队列是按照优先级级顺序排序的，优先级大的在队头。队列中的对象应该实现Comparable接口。在compareTo中，当和其他对象比较时，如果该方法返回负数，那么在队列里面的优先级就比较高。

package com.dy.xidian;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    private Random rand = new Random(47);
    private static int counter = 0;
    private final int id = counter++;
    private final int priority;

    protected static List<PrioritizedTask> sequeue = new ArrayList<PrioritizedTask>();

    public PrioritizedTask(int priority) {
        super();
        this.priority = priority;
        sequeue.add(this);
    }

    @Override
    public int compareTo(PrioritizedTask that) {
        if (this.priority > that.priority)
            return -1;
        if (this.priority < that.priority)
            return 1;
        return 0;
    }

    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(rand.nextInt(250));
            System.out.println(this);
        } catch (InterruptedException e) {
        }

    }

    @Override
    public String toString() {
        return String.format("[%1$-3d]", priority) + "Task" + id;
    }

    public String summary() {
        return "(" + id + " : " + priority + ")";
    }

    public static class EndSentinel extends PrioritizedTask {
        private ExecutorService exec;

        public EndSentinel(ExecutorService e) {
            super(-1);
            exec = e;
        }

        public void run() {
            int count = 0;
            for (PrioritizedTask pt : sequeue) {
                System.out.println(pt.summary());
                if (++count % 5 == 0)
                    System.out.println("");
            }
            System.out.println("");
            System.out.println(this + " Calling shutdownNow()");
            exec.shutdownNow();
        }
    }
}

class PrioritizedTaskProducer implements Runnable {
    private Random rand = new Random(47);
    private Queue<Runnable> queue;
    private ExecutorService exec;

    public PrioritizedTaskProducer(Queue<Runnable> queue, ExecutorService exec) {
        super();
        this.queue = queue;
        this.exec = exec;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            queue.add(new PrioritizedTask(rand.nextInt(10)));
            Thread.yield();
        }
        try {
            for (int i = 0; i < 10; i++) {
                TimeUnit.MILLISECONDS.sleep(250);
                queue.add(new PrioritizedTask(10));
            }

        } catch (InterruptedException e) {
        }
        for (int i = 0; i < 10; i++)
            queue.add(new PrioritizedTask(i));
        queue.add(new PrioritizedTask.EndSentinel(exec));
        System.out.println("Finished PrioritizedTaskProducer");
    }
}

class PrioritizedTaskConsumer implements Runnable {
    private PriorityBlockingQueue<Runnable> q;
    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
        super();
        this.q = q;
    }

    @Override
    public void run() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e1) {
        }
        try {
            while (!Thread.interrupted()) {
                q.take().run();
            }
        } catch (InterruptedException e) {
            System.out.println("Interrupted Execption!");
        }
        System.out.println("Finished PrioritizedTaskConsumer");
    }
}

public class PriorityBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<Runnable>();
        exec.execute(new PrioritizedTaskProducer(queue, exec));
        exec.execute(new PrioritizedTaskConsumer(queue));
    }
}

4.ScheduledThreadPoolExecutor

制定任务计划表，指定主线程运行多少秒(毫秒)后，开启子线程来运行别的任务。

package com.dy.xidian;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TaskTest {
    static ScheduledThreadPoolExecutor scheduler = null;
    static int index = 0;

    public static void main(String[] args) {

        // 构造一个ScheduledThreadPoolExecutor对象，并且设置它的容量为5个
        scheduler = new ScheduledThreadPoolExecutor(5);
        MyTask task = new MyTask();
        // 隔2秒后开始执行任务，并且在上一次任务开始后隔一秒再执行一次；
        // stpe.scheduleWithFixedDelay(task, 2, 1, TimeUnit.SECONDS);
        // 隔6秒后执行一次，但只会执行一次。
        for (int i = 0; i < 10; i++)
            scheduler.schedule(task, i + 1, TimeUnit.SECONDS);
    }

    private static String getTimes() {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss E");
        Date date = new Date();
        date.setTime(System.currentTimeMillis());
        return format.format(date);
    }

    private static class MyTask implements Runnable {

        @Override
        public void run() {
            index++;
            System.out.println(getTimes() + " " + index);
            if (index >= 10) {
                scheduler.shutdownNow();
            }
        }
    }
}

5.Semaphore

　　同类资源只有一个的话，我们可以用Lock或者是synchronized来对它进行互斥访问。当同类资源数量有多个，能够满足多个线程同时操作时，可以考虑到使用信号量来实现互斥访问。

对象池代码

package com.dy.xidian;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;

public class ObjectPool<T> {
    private int size;
    private List<T> items = new ArrayList<T>();
    private volatile boolean[] checkOut;
    private Semaphore available;

    public ObjectPool(Class<T> classObject, int size) {
        this.size = size;
        checkOut = new boolean[size];
        //size表示初始资源数,true表示对请求进行先来先服务操作
        available = new Semaphore(size, true);
        for (int i = 0; i < size; i++) {
            try {
                items.add(classObject.newInstance());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    public T checkOut() throws InterruptedException {
        //获取信号量,如果没有资源请等待,信号量计数减1
        available.acquire();
        return getItem();
    }

    private synchronized T getItem() {
        for (int i = 0; i < size; i++) {
            if (!checkOut[i]) {
                checkOut[i] = true;
                return items.get(i);
            }
        }
        return null;
    }

    public void checkIn(T x) {
        //归还资源，释放信号量,信号量计数加1
        if (releaseItem(x))
            available.release();
    }

    private synchronized boolean releaseItem(T item) {
        int index = items.indexOf(item);
        if (index == -1)
            return false;
        if (checkOut[index]) {
            checkOut[index] = false;
            return true;
        }
        return false;
    }
}

信号量Demo:

package com.dy.xidian;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

class Fat {
    private volatile double d = 0;
    private static int counter = 0;
    private final int id = counter++;
    public Fat() {
        for (int i = 0; i < 1000; i++) {
            d += (Math.PI + Math.E) / (double) i;
        }
    }

    public void operation() {
        System.out.println(this);
    }

    public String toString() {
        return "Fat id: " + id;
    }
}
class CheckOutTask<T> implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private ObjectPool<T> pool;

    public CheckOutTask(ObjectPool<T> pool) {
        this.pool = pool;
    }

    @Override
    public void run() {
        try {
            T item = pool.checkOut();
            System.out.println(this + "checked out " + item);
            TimeUnit.SECONDS.sleep(1);
            System.out.println(this + "checked in " + item);
            pool.checkIn(item);
        } catch (InterruptedException e) {
        }
    }

    public String toString() {
        return "CheckoutTask" + id + " ";
    }
}

public class SemaphoreDemo {
    final static int SIZE = 10;

    public static void main(String[] args) throws InterruptedException {
        final ObjectPool<Fat> pool = new ObjectPool<Fat>(Fat.class, SIZE);
        ExecutorService exec = Executors.newCachedThreadPool();
        List<Fat> list = new ArrayList<Fat>();
        // 创建10个子线程进行签入、签出操作
        for (int i = 0; i < SIZE; i++)
            exec.execute(new CheckOutTask<Fat>(pool));
        System.out.println("All checkout Task created");

        // 主线程只签出
        for (int i = 0; i < SIZE; i++) {
            Fat f = pool.checkOut();
            System.out.println(i + " : main() thread checked out");
            f.operation();
            list.add(f);
        }

        Future<?> blocked = exec.submit(new Runnable() {

            @Override
            public void run() {
                try {
                    pool.checkOut();
                } catch (InterruptedException e) {
                    System.out.println("checkOut() interrupted");
                }
            }
        });
        TimeUnit.SECONDS.sleep(2);
        blocked.cancel(true);
        System.out.println("Checking in objects in " + list);
        for (Fat fat : list)
            pool.checkIn(fat);
        for (Fat fat : list)
            pool.checkIn(fat);
        exec.shutdown();
    }
}

　　代码中创建了一个对象池，每次从池中获取对象时都要先获取信号量,如果信号量计数小于或等于0,则等待。信号量在创建时需要指定资源数，对象池中最开始有10个对象，则信号量的初始值应为10。在使用完对象后应该归还该对象并释放信号量。对于信号量的使用比较简答些。关于对象池代码则可以作为今后编程中示例代码。

6.Exchanger

　　Exchanger可以在两个线程之间交换数据，只能是2个线程，他不支持更多的线程之间互换数据。当线程A调用Exchange对象的exchange()方法后，他会陷入阻塞状态，直到线程B也调用了exchange()方法，然后以线程安全的方式交换数据，之后线程A和B继续运行。

package com.dy.xidian;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import net.mindview.util.BasicGenerator;
import net.mindview.util.Generator;

class ExchangerProducer<T> implements Runnable {
    private Generator<T> generator;
    private Exchanger<List<T>> exchanger;
    private List<T> holder;

    /**
     * 
     * @param exchanger交换器
     *            , 用于交换对象
     * @param generator产生器
     *            , 产生要交换的数据
     * @param holder数据容器
     *            , 用来存储产生的数据
     */
    public ExchangerProducer(Exchanger<List<T>> exchanger,
            Generator<T> generator, List<T> holder) {
        super();
        this.generator = generator;
        this.exchanger = exchanger;
        this.holder = holder;
    }

    /**
     * 生产者线程会生成一个满的List,用于交换对象
     */
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                for (int i = 0; i < ExchangerDemo.size; i++)
                    holder.add(generator.next());
                // 返回值是从消费者那里拿到的数据(其实就是一个空表)
                holder = exchanger.exchange(holder);
            }
        } catch (InterruptedException e) {
        }
    }
}

class ExchangerConsumer<T> implements Runnable {
    private Exchanger<List<T>> exchanger;
    private List<T> holder;
    private volatile T value;

    /**
     * 
     * @param exchanger交换器
     *            ,用于交换数据
     * @param holder
     *            欲交换的对象
     */
    public ExchangerConsumer(Exchanger<List<T>> exchanger, List<T> holder) {
        super();
        this.exchanger = exchanger;
        this.holder = holder;
    }

    /**
     * 消费者不断将表中的元素移除,给生产者一个空表
     */
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                holder = exchanger.exchange(holder);
                for (T x : holder) {
                    value = x;
                    holder.remove(x);
                }
            }
        } catch (InterruptedException e) {
        }
        System.out.println("Final value: " + value);
    }
}

public class ExchangerDemo {
    static int size = 10;
    static int delay = 5;

    public static void main(String[] args) throws Exception {
        if (args.length > 0)
            size = new Integer(args[0]);
        if (args.length > 1)
            delay = new Integer(args[1]);
        ExecutorService exec = Executors.newCachedThreadPool();
        Exchanger<List<Fat>> xc = new Exchanger<List<Fat>>();
        List<Fat> producerList = new CopyOnWriteArrayList<Fat>(), consumerList = new CopyOnWriteArrayList<Fat>();
        exec.execute(new ExchangerProducer<Fat>(xc, BasicGenerator
                .create(Fat.class), producerList));
        exec.execute(new ExchangerConsumer<Fat>(xc, consumerList));
        TimeUnit.SECONDS.sleep(delay);
        exec.shutdownNow();
    }
}

代码中ExchangeProducer不断填充List,然后将这个满表交换为ExchangerConsumer传递给它的空表。

代码中用到的写时拷贝技术可以查考下面链接

http://www.cnblogs.com/dolphin0520/p/3938914.html