//自定义的缓存public class SemaphoreBoundedBuffer <E> {private final Semaphore availableItems, availableSpaces;private final E[] items;private int putPosition = 0, takePosition = 0;public SemaphoreBoundedBuffer(int capacity) {if (capacity <= 0)throw new IllegalArgumentException();
        availableItems = new Semaphore(0);
        availableSpaces = new Semaphore(capacity);
        items = (E[]) new Object[capacity];
    }public boolean isEmpty() {return availableItems.availablePermits() == 0;
    }public boolean isFull() {return availableSpaces.availablePermits() == 0;
    }public void put(E x) throws InterruptedException {
        availableSpaces.acquire();
        doInsert(x);
        availableItems.release();
    }public E take() throws InterruptedException {
        availableItems.acquire();
        E item = doExtract();
        availableSpaces.release();return item;
    }private synchronized void doInsert(E x) {int i = putPosition;
        items[i] = x;
        putPosition = (++i == items.length) ? 0 : i;
    }private synchronized E doExtract() {int i = takePosition;
        E x = items[i];
        items[i] = null;
        takePosition = (++i == items.length) ? 0 : i;return x;
    }
}//无需比较每次 生产者和消费者取出的值；只需要最终比较和即可public class PutTakeTest extends TestCase {protected static final ExecutorService pool = Executors.newCachedThreadPool();protected CyclicBarrier barrier;protected final SemaphoreBoundedBuffer<Integer> bb;protected final int nTrials, nPairs;protected final AtomicInteger putSum = new AtomicInteger(0);protected final AtomicInteger takeSum = new AtomicInteger(0);public static void main(String[] args) throws Exception {new PutTakeTest(10, 10, 100000).test(); // sample parameters        pool.shutdown();
    }public PutTakeTest(int capacity, int npairs, int ntrials) {this.bb = new SemaphoreBoundedBuffer<Integer>(capacity);this.nTrials = ntrials;this.nPairs = npairs;this.barrier = new CyclicBarrier(npairs * 2 + 1);   //初始化为 线程数量*2 + 1：生产者+消费者+主线程    }void test() {try {for (int i = 0; i < nPairs; i++) {
                pool.execute(new Producer());
                pool.execute(new Consumer());
            }
            barrier.await(); // 等待所有线程初始化完成，完成后复位栅栏barrier.await(); // 等待所有线程执行完成assertEquals(putSum.get(), takeSum.get());      //执行比较，判断线程安全性能} catch (Exception e) {throw new RuntimeException(e);
        }
    }static int xorShift(int y) {    //生成随机数y ^= (y << 6);
        y ^= (y >>> 21);
        y ^= (y << 7);return y;
    }class Producer implements Runnable {public void run() {try {int seed = (this.hashCode() ^ (int) System.nanoTime());int sum = 0;
                barrier.await();                        //等待所有线程初始化完成for (int i = nTrials; i > 0; --i) {
                    bb.put(seed);
                    sum += seed;
                    seed = xorShift(seed);
                }
                putSum.getAndAdd(sum);
                barrier.await();                        //等待所有线程执行完成} catch (Exception e) {throw new RuntimeException(e);
            }
        }
    }class Consumer implements Runnable {public void run() {try {
                barrier.await();                        //等待所有线程初始化完成int sum = 0;for (int i = nTrials; i > 0; --i) {
                    sum += bb.take();
                }
                takeSum.getAndAdd(sum);
                barrier.await();                        //等待所有线程执行完成} catch (Exception e) {throw new RuntimeException(e);
            }
        }
    }
}