快速上手JUC下常见并发容器

/** The queued items */
final Object[] items;

// 这是一个掌管所有访问操作的锁。全局共享。都会使用这个锁。
final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;

    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock; // 唯一锁
        lock.lockInterruptibly();// 加锁
        try {
            while (count == items.length)
                notFull.await();//await 让出操作权
            enqueue(e);// 被唤醒就加入队列。
        } finally {
            lock.unlock();// 解锁
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock; // 加锁
        lock.lockInterruptibly();
        try {
            while (count == 0)//为空则释放当前锁
                notEmpty.await();
            return dequeue();// 获得锁被唤醒了则返回数据
        } finally {
            lock.unlock();// 释放锁
        }
    }

//节点类，用于存储数据
static class Node<E> {
    E item;
    Node<E> next;
    Node(E x) { item = x; }
}

// 阻塞队列的大小，默认为Integer.MAX_VALUE 
private final int capacity;
//当前阻塞队列中的元素个数 
private final AtomicInteger count = new AtomicInteger();
// 阻塞队列的头结点
transient Node<E> head;
// 阻塞队列的尾节点
private transient Node<E> last;
// 获取并移除元素时使用的锁，如take, poll
private final ReentrantLock takeLock = new ReentrantLock();
//  notEmpty条件对象，当队列没有数据时用于 挂起 执行删除的线程
private final Condition notEmpty = takeLock.newCondition();
// 添加元素时使用的锁如 put, offer
private final ReentrantLock putLock = new ReentrantLock();
// notFull条件对象，当队列数据已满时用于 挂起 执行添加的线程
private final Condition notFull = putLock.newCondition();

public interface TransferQueue<E> extends BlockingQueue<E> {
    // 如果可能，立即将元素转移给等待的消费者。 
    // 更确切地说，如果存在消费者已经等待接收它（在 take 或 timed poll（long，TimeUnit）poll）中，则 立即传送指定 的元素，否则返回 false。
    boolean tryTransfer(E e);

    // 将元素转移给消费者，如果需要的话等待。 
    // 更准确地说，如果存在一个消费者已经等待接收它（在 take 或timed poll（long，TimeUnit）poll）中，则立即传送指定的元素，否则 等待 直到 元素由消费者接收。
    void transfer(E e) throws InterruptedException;

    // 上面方法的基础上设置超时时间
    boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;

    // 如果至少有一位消费者在等待，则返回 true
    boolean hasWaitingConsumer();

    // 返回等待消费者人数的估计值
    int getWaitingConsumerCount();
}

  public boolean add(E e) {
        addLast(e); // 等价
        return true;
    }

    public E remove() {
        return removeFirst();
    }

    public E take() throws InterruptedException {
        return takeFirst();
    }

public class Order {
 private final String orderNo;//订单的编号
 private final double orderMoney;//订单的金额
 public Order(String orderNo, double orderMoney) {
  super();
  this.orderNo = orderNo;
  this.orderMoney = orderMoney;
 }
 public String getOrderNo() {
  return orderNo;
 }
 public double getOrderMoney() {
  return orderMoney;
 }
}

// 类说明：存放到队列的元素
public class ItemVo<T> implements Delayed {

    private long activeTime;//到期时间，单位毫秒
    private T object;

    //activeTime是个过期时长
    public ItemVo(long activeTime, T object) {
        super();
        this.activeTime = TimeUnit.NANOSECONDS.convert(activeTime, TimeUnit.MILLISECONDS) + System.nanoTime();
        // 将传入的时长转换为超时的时刻
        this.object = object;
    }

    public T getObject() {
        return object;
    }

    //按照剩余时间排序
    @Override
    public int compareTo(Delayed o) {
        long d = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
        return (d == 0) ? 0 : ((d > 0) ? 1 : -1);
    }

    //返回元素的剩余时间
    @Override
    public long getDelay(TimeUnit unit) {
        long d = unit.convert(this.activeTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        return d;
    }
}

public class PutOrder implements Runnable {
    private DelayQueue<ItemVo<Order>> queue;
    public PutOrder(DelayQueue<ItemVo<Order>> queue) {
        super();
        this.queue = queue;
    }
    @Override
    public void run() {
        //5秒到期
        Order ordeTb = new Order("TBSoWhat", 14);
        ItemVo<Order> itemTb = new ItemVo<Order>(5000, ordeTb);
        queue.offer(itemTb); //插入
        System.out.println("订单5秒后到期：" + ordeTb.getOrderNo());
        //8秒到期
        Order ordeJd = new Order("JDSoWhat", 12);
        ItemVo<Order> itemJd = new ItemVo<Order>(8000, ordeJd);
        queue.offer(itemJd);// 插入
        System.out.println("订单8秒后到期：" + ordeJd.getOrderNo());
    }
}

public class FetchOrder implements Runnable {
 private DelayQueue<ItemVo<Order>> queue;
 public FetchOrder(DelayQueue<ItemVo<Order>> queue) {
  super();
  this.queue = queue;
 }
 @Override
 public void run() {
  while(true) {
   try {
    ItemVo<Order> item = queue.take();
    Order order = (Order)item.getObject();
    System.out.println("get from queue:"+order.getOrderNo());
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
 } 
}

public class Test
{
 public static void main(String[] args) throws InterruptedException
 {
  DelayQueue<ItemVo<Order>> queue = new DelayQueue<>();
  new Thread(new PutOrder(queue)).start();
  new Thread(new FetchOrder(queue)).start();
  //每隔1秒，打印个数字
  for (int i = 1; i < 10; i++)
  {
   Thread.sleep(1000);
   System.out.println(i * 1000);
  }
 }
}

public class ForkJoinWork extends RecursiveTask<Long> {
    private Long start;//起始值
    private Long end;//结束值
    public static final Long critical = 100000L;//临界值

    public ForkJoinWork(Long start, Long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // return null;
        //判断是否是拆分完毕
        Long lenth = end - start;   //起始值差值
        if (lenth <= critical) {
            //如果拆分完毕就相加
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else {
            //没有拆分完毕就开始拆分
            Long middle = (end + start) / 2;//计算的两个值的中间值
            ForkJoinWork right = new ForkJoinWork(start, middle);
            right.fork();//拆分，并压入线程队列
            ForkJoinWork left = new ForkJoinWork(middle + 1, end);
            left.fork();//拆分，并压入线程队列

            //合并
            return right.join() + left.join();
        }

    }
}

public class ForkJoinWorkTest {
    @Test
    public void test() {
        //ForkJoin实现
        long l = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();//实现ForkJoin 就必须有ForkJoinPool的支持
        ForkJoinTask<Long> task = new ForkJoinWork(0L, 10000000000L);//参数为起始值与结束值
        Long invoke = forkJoinPool.invoke(task);
        long l1 = System.currentTimeMillis();
        System.out.println("invoke = " + invoke + "  time: " + (l1 - l));
        //invoke = -5340232216128654848  time: 56418
        //ForkJoinWork forkJoinWork = new ForkJoinWork(0L, 10000000000L);
    }
    @Test
    public void test2() {
        //普通线程实现
        Long x = 0L;
        Long y = 10000000000L;
        long l = System.currentTimeMillis();
        for (Long i = 0L; i <= y; i++) {
            x += i;
        }
        long l1 = System.currentTimeMillis();
        System.out.println("invoke = " + x + "  time: " + (l1 - l));
        //invoke = -5340232216128654848  time: 64069
    }

    @Test
    public void test3() {
        //Java 8 并行流的实现
        long l = System.currentTimeMillis();
        long reduce = LongStream.rangeClosed(0, 10000000000L).parallel().reduce(0, Long::sum);
        long l1 = System.currentTimeMillis();
        System.out.println("invoke = " + reduce + "  time: " + (l1 - l));
        //invoke = -5340232216128654848  time: 2152
    }
}

public class ForkJoinWorkerThread extends Thread {
    
    final ForkJoinPool pool;                    // 该工作线程归属的线程池
    final ForkJoinPool.WorkQueue workQueue;     // 对应的任务队列
 
    protected ForkJoinWorkerThread(ForkJoinPool pool) {
        super("aForkJoinWorkerThread");         // 指定工作线程名称
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }
  
    // ...
}

volatile WorkQueue[] workQueues; // main registry

CountDownLatch(int count) //实例化一个倒计数器，count指定计数个数
countDown() // 计数减一
await() //等待，当计数减到0时，所有线程并行执行

/**
 * @Description: 倒计时器示例:火箭发射
 */
public class CountDownLatchDemo implements Runnable{
    static final CountDownLatch latch = new CountDownLatch(10);
    static final CountDownLatchDemo demo = new CountDownLatchDemo();

    @Override
    public void run() {
        // 模拟检查任务
        try {
            Thread.sleep(new Random().nextInt(10) * 1000);
            System.out.println("检查完毕");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //计数减一
            //放在finally避免任务执行过程出现异常，导致countDown()不能被执行
            latch.countDown();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newFixedThreadPool(10);
        for (int i=0; i<10; i++){
            exec.submit(demo);
        }
        // 等待检查
        latch.await(); // 外部主线程main 方法来等待下面运行！！！
        // 发射火箭
        System.out.println("Fire!");
        // 关闭线程池
        exec.shutdown();
    }
}

 for (int i = 0; i < 10; i++) {
     countDownLatch.countDown();
 }

public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)

public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException

public class CyclicBarrierTest {
    public static void main(String[] args) {
        //定义一个计数器，当计数器的值累加到30，输出"放行"
        CyclicBarrier cyclicBarrier = new CyclicBarrier(30,()->{
            System.out.println("放行");
        });
        for (int i = 1; i <= 90; i++) {
            final int temp = i;
            new Thread(()->{
                System.out.println("-->"+temp);
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();

        }
    }
}

public class SemaphoreTest {
    public static void main(String[] args) {
        //同时只能进5个人
        Semaphore semaphore = new Semaphore(5);
        for (int i = 0; i < 15; i++) {
            new Thread(() -> {
                try {
                    //获得许可
                    semaphore.acquire(); // 已经进店人+1
                    System.out.println(Thread.currentThread().getName() + "进店购物");
                    TimeUnit.SECONDS.sleep(5);
                    System.out.println(Thread.currentThread().getName() + "出店");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    //释放许可
                    semaphore.release(); //已经进店人 -1 
                }
            }, String.valueOf(i)).start();
        }
    }
}

public class SqlConnectImpl implements Connection{
 
 /*拿一个数据库连接*/
    public static final Connection fetchConnection(){
        return new SqlConnectImpl();
    }
}

public class DBPoolSemaphore {

    private final static int POOL_SIZE = 10;
    private final Semaphore useful, useless;//useful表示可用的数据库连接，useless表示已用的数据库连接

    public DBPoolSemaphore() {
        this.useful = new Semaphore(POOL_SIZE);
        this.useless = new Semaphore(0);
    }

    //存放数据库连接的容器
    private static LinkedList<Connection> pool = new LinkedList<Connection>();

    //初始化池
    static {
        for (int i = 0; i < POOL_SIZE; i++) {
            pool.addLast(SqlConnectImpl.fetchConnection());
        }
    }

    /*归还连接*/
    public void returnConnect(Connection connection) throws InterruptedException {
        if (connection != null) {
            System.out.println("当前有" + useful.getQueueLength() + "个线程等待数据库连接！！"
                    + "可用连接数:" + useful.availablePermits());
            useless.acquire();// 可用连接 +1
            synchronized (pool) {
                pool.addLast(connection);
            }
            useful.release(); // 已用连接 -1
        }
    }

    /*从池子拿连接*/
    public Connection takeConnect() throws InterruptedException {
        useful.acquire(); // 可用连接-1
        Connection conn;
        synchronized (pool) {
            conn = pool.removeFirst();
        }
        useless.release(); // 以用连接+1
        return conn;
    }
}

public class AppTest {

    private static DBPoolSemaphore dbPool = new DBPoolSemaphore();

    //业务线程
    private static class BusiThread extends Thread {
        @Override
        public void run() {
            Random r = new Random();//让每个线程持有连接的时间不一样
            long start = System.currentTimeMillis();
            try {
                Connection connect = dbPool.takeConnect();
                System.out.println("Thread_" + Thread.currentThread().getId()
                        + "_获取数据库连接共耗时【" + (System.currentTimeMillis() - start) + "】ms.");
                SleepTools.ms(100 + r.nextInt(100));//模拟业务操作，线程持有连接查询数据
                System.out.println("查询数据完成，归还连接！");
                dbPool.returnConnect(connect);
            } catch (InterruptedException e) {
            }
        }
    }

    public static void main(String[] args) {
        for (int i = 0; i < 50; i++) {
            Thread thread = new BusiThread();
            thread.start();
        }
    }
}

public class UseExchange {
    private static final Exchanger<Set<String>> exchange
            = new Exchanger<Set<String>>();

    public static void main(String[] args) {

        //第一个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                Set<String> setA = new HashSet<String>();//存放数据的容器
                try {
                    setA.add("liu");
                    setA.add("Liu");
                    setA.add("LIU");
                    setA = exchange.exchange(setA);//交换set
                    /*处理交换后的数据*/
                } catch (InterruptedException e) {
                }
            }
        }).start();

        //第二个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                Set<String> setB = new HashSet<String>();//存放数据的容器
                try {
                    setB.add("jin");
                    setB.add("Jie");
                    setB.add("JIN");
                    setB = exchange.exchange(setB);//交换set
                    /*处理交换后的数据*/
                } catch (InterruptedException e) {
                }
            }
        }).start();
    }
}

public class UseFuture {

    /*实现Callable接口，允许有返回值*/
    private static class UseCallable implements Callable<Integer> {
        private int sum;
        @Override
        public Integer call() throws Exception 
            System.out.println("Callable子线程开始计算");
            Thread.sleep(2000);
            for (int i = 0; i < 5000; i++) {
                sum = sum + i;
            }
            System.out.println("Callable子线程计算完成，结果=" + sum);
            return sum;
        }
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        UseCallable useCallable = new UseCallable();
        FutureTask<Integer> futureTask = new FutureTask<Integer>(useCallable);
        new Thread(futureTask).start();
        Random r = new Random();
        SleepTools.second(1);
        if (r.nextBoolean()) {  // 方法调用返回下一个伪均匀分布的boolean值
            System.out.println("Get UseCallable result = " + futureTask.get());
        } else {
            System.out.println("中断计算");
            futureTask.cancel(true);
        }
    }
}