【JDK并发包基础】并发容器详解

    public static void main(String[] args) {
		ConcurrentHashMap<String, Object> chm = new ConcurrentHashMap<String, Object>();
		chm.put("k1", "v1");
		chm.put("k2", "v2");
		chm.put("k3", "v3");
		chm.putIfAbsent("k3", "vvvv");
		System.out.println(chm.get("k2"));
		System.out.println(chm.size());
		for(Map.Entry<String, Object> me : chm.entrySet()){
			System.out.println("key:" + me.getKey() + ",value:" + me.getValue());
		}
	}

   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();
        }
    }

     private final CopyOnWriteArrayList<E> al;
     public CopyOnWriteArraySet() {
        al = new CopyOnWriteArrayList<E>();
    }

		//高性能无阻塞无界队列：ConcurrentLinkedQueue
		ConcurrentLinkedQueue<String> qlq = new ConcurrentLinkedQueue<String>();
		qlq.offer("a");
		qlq.offer("b");
		qlq.offer("c");
		qlq.offer("d");
		qlq.add("e");
		
		System.out.println(qlq.poll());	//a 从头部取出元素，并从队列里删除
		System.out.println(qlq.size());	//4
		System.out.println(qlq.peek());	//b
		System.out.println(qlq.size());	//4

		ArrayBlockingQueue<String> array = new ArrayBlockingQueue<String>(5);
		array.put("a");
		array.put("b");
		array.add("c");
		array.add("d");
		array.add("e");
		//array.add("f");这里放开会报异常,因为已经添加了5个元素了
        //2秒后返回false,表示添加失败
		System.out.println(array.offer("a", 2, TimeUnit.SECONDS));

public class BlockingQueueHosee {
	//数据生产器
	private static AtomicInteger count = new AtomicInteger();
	
	final BlockingQueue<Integer> bq = new LinkedBlockingQueue<Integer>(10);
	
	class Producer implements Runnable {
		//多线程间是否启动变量，有强制从主内存中刷新的功能。即时返回线程的状态
		private volatile boolean isRunning = true;
		//随机对象
		private  Random r = new Random(); 
		@Override
		public void run() {
			while(isRunning){
				try {
					//随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时) 
					Thread.sleep(r.nextInt(1000));
					//获取的数据进行累计..
					int incrementCount = count.incrementAndGet();
					bq.put(incrementCount);
					System.out.println(
							"当前线程:" + Thread.currentThread().getName() + ", 生产了数据"+incrementCount+", 进行装载到公共缓冲区中..."
							);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		public void stop(){
			this.isRunning = false;
		}
	}

	class Consumer implements Runnable {
		//随机对象
		private  Random r = new Random(); 
		@Override
		public void run() {
			while(true){
				try {
					Integer take = bq.take();
					//进行数据处理。休眠0 - 1000毫秒模拟耗时
					Thread.sleep(r.nextInt(1000));
					System.out.println("当前消费线程：" + Thread.currentThread().getName() + "， 消费成功，消费数据为id: " + take);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static void main(String[] args) throws Exception {
		BlockingQueueHosee hosee = new BlockingQueueHosee();
		Producer producer = hosee.new Producer();
		new Thread(producer,"生产者").start();
		new Thread(hosee.new Consumer(),"消费者1").start();
		new Thread(hosee.new Consumer(),"消费者2").start();
		Thread.sleep(3000);
		producer.stop();
	}
}

//需要先取数据,才能向SynchronousQueue添加数据,直接添加会报错
final SynchronousQueue<String> q = new SynchronousQueue<String>();
		//q.add("adsa ");// java.lang.IllegalStateException: Queue full
		Thread t1 = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					System.out.println(q.take());
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});
		t1.start();
		Thread t2 = new Thread(new Runnable() {
			@Override
			public void run() {
				q.add("asdasd");
			}
		});
		t2.start();		
	}