08 容器的魔力：并发世界的宝库

  public synchronized boolean add(E e) {
      ...
  }

    public synchronized V put(K key, V value) {
        .....
    }

   List<String> list = Collections.synchronizedList(new ArrayList<>());

   public boolean add(E e) {
       synchronized (mutex) {return c.add(e);}
   }

public class NewSyncContainer {
    public static void main(String[] args) {
        ConcurrentHashMap<String,String> concurrentHashMap = new ConcurrentHashMap<>();
        concurrentHashMap.put("1","a");
    }
}

public static void main(String[] args) {
    CopyOnWriteArrayList<String> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
    copyOnWriteArrayList.add("a");
    copyOnWriteArrayList.add("b");
    for (String s : copyOnWriteArrayList) {
        System.out.println(s);
    }
}

public boolean add(E e) {
    //增加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
        // 复制一个集合的副本
        Object[] newElements = Arrays.copyOf(elements, len + 1);
        // 将要增加的线程追加到副本中
        newElements[len] = e;
        // 使用副本替换集合
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

public class BlockingQueueTest {
    private final static ArrayBlockingQueue<String> ARRAY_BLOCKING_QUEUE = new ArrayBlockingQueue<String>(5);
    private final static ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        //触发排队操作  这里实际生产环境不推荐两种任务共同使用一个线程池 演示使用
        THREAD_POOL_EXECUTOR.execute(new QueueUp());

        //三个服务员开始接待客人吃饭
        for (int i = 0; i < 3; i++) {
            THREAD_POOL_EXECUTOR.execute(new EatingTask());
        }
    }

    private static class QueueUp implements Runnable {

        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    String customerName = "顾客" + i;
                    System.out.println("顾客" + customerName + "开始排队");
                    ARRAY_BLOCKING_QUEUE.put(customerName);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static class EatingTask implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    String thisCustomerName = ARRAY_BLOCKING_QUEUE.take();
                    System.out.println("顾客" + thisCustomerName + "排队成功，进入餐厅开始吃饭");
                    Thread.sleep((long) (Math.random() * 10000));
                    System.out.println("顾客" + thisCustomerName + "吃完离开了.");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

        }
    }
}

  public void put(E e) throws InterruptedException {
      checkNotNull(e);
      final ReentrantLock lock = this.lock;
      // 进来先加锁
      lock.lockInterruptibly();
      try {
          //判断队列是否已经满了
          while (count == items.length)
              //满了就进行阻塞
              notFull.await();
          //队列没满就入队
          enqueue(e);
      } finally {
          //释放锁
          lock.unlock();
      }
  }

public class PriorityQueueTest {

    private final static PriorityBlockingQueue<Integer> priorityQueue = new PriorityBlockingQueue<>(2, new Comparator<Integer>() {
        @Override
        public int compare(Integer o1, Integer o2) {
            //倒序排列
            return o2 - o1;
        }
    });

    public static void main(String[] args) throws InterruptedException {
        priorityQueue.add(1);
        priorityQueue.add(2);
        priorityQueue.add(3);
        priorityQueue.add(4);
        //4
        System.out.println(priorityQueue.take());
        //3
        System.out.println(priorityQueue.take());
        //2
        System.out.println(priorityQueue.take());
        //1
        System.out.println(priorityQueue.take());
    }
}

4
3
2
1

public class SynchronousQueueTest {

    public static void main(String[] args) throws InterruptedException {
        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

        //模拟生产者
        new Thread(() ->{
            while (true) {
                try {
                    synchronousQueue.put("你好");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();


        //模拟消费者
        new Thread(() ->{
            while (true) {
                try {
                    //睡眠 使消费者的消费速度低于数据产生的速度
                    Thread.sleep(1000);
                    System.out.println(synchronousQueue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }
}

public class DelayQueueTest {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedTask> delayQueue = new DelayQueue<DelayedTask>();
        delayQueue.add(new DelayedTask("a1", 3000));
        delayQueue.add(new DelayedTask("11", 3000));
        delayQueue.add(new DelayedTask("2", 4000));
        delayQueue.add(new DelayedTask("3", 5000));

        System.out.println(delayQueue.take().taskName);
        System.out.println(delayQueue.take().taskName);
        System.out.println(delayQueue.take().taskName);
        System.out.println(delayQueue.take().taskName);
    }

    private static class DelayedTask implements Delayed {
        private long delayTime;  // 延迟时间，单位为纳秒
        private String taskName;

        public DelayedTask(String taskName, long delayTime) {
            this.taskName = taskName;
            this.delayTime = System.currentTimeMillis() + delayTime;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            long diff = delayTime - System.currentTimeMillis();
            return unit.convert(diff, TimeUnit.MILLISECONDS);
        }

        /**
         * 用于延迟队列内部比较排序
         **/
        @Override
        public int compareTo(Delayed o) {
            long diff = this.delayTime - ((DelayedTask) o).delayTime;
            return Long.compare(diff, 0);
        }

        public String getTaskName() {
            return taskName;
        }
    }

}

public class ProducerConsumerExample {
    private static final int MAX_CAPACITY = 5;
    private static ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();

    public static void main(String[] args) {
        Thread producer = new Thread(new Producer());
        Thread consumer = new Thread(new Consumer());

        producer.start();
        consumer.start();
    }

    static class Producer implements Runnable {
        @Override
        public void run() {
            while (true) {
                if (queue.size() == MAX_CAPACITY) {
                    System.out.println("队列已满，等待消费者消费...");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    int number = (int) (Math.random() * 100);
                    queue.offer(number);
                    System.out.println("生产者生产的消息为: " + number);
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                if (queue.isEmpty()) {
                    System.out.println("队列为空，等待生产者生产...");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    int number = queue.poll();
                    System.out.println("消费者消费到数据: " + number);
                }
            }
        }
    }
}

队列为空，等待生产者生产...
生产者生产的消息为: 85
生产者生产的消息为: 12
生产者生产的消息为: 47
生产者生产的消息为: 84
生产者生产的消息为: 72
队列已满，等待消费者消费...
队列已满，等待消费者消费...
消费者消费到数据: 85
消费者消费到数据: 12
消费者消费到数据: 47
消费者消费到数据: 84
消费者消费到数据: 72
队列为空，等待生产者生产...
生产者生产的消息为: 31
生产者生产的消息为: 9
生产者生产的消息为: 11
生产者生产的消息为: 20
生产者生产的消息为: 15
队列已满，等待消费者消费...