高并发编程基础（java.util.concurrent包常见类基础）

2022-11-12 12:01:40

　　JDK5中添加了新的java.util.concurrent包，相对同步容器而言，并发容器通过一些机制改进了并发性能。因为同步容器将所有对容器状态的访问都串行化了，这样保证了线程的安全性，所以这种方法的代价就是严重降低了并发性，当多个线程竞争容器时，吞吐量严重降低。因此JDK5开始针对多线程并发访问设计，提供了并发性能较好的并发容器，引入了java.util.concurrent包。与Vector和Hashtable、Collections.synchronizedXxx()同步容器等相比，util.concurrent中引入的并发容器主要解决了两个问题：

1）根据具体场景进行设计，尽量避免synchronized，提供并发性。
2）定义了一些并发安全的复合操作，并且保证并发环境下的迭代操作不会出错。

这里先粗略的介绍下 concurrrent 包下面常见并发容器的简单应用，后续再针对性的去深入研究。

并发容器：

这些容器的关键方法大部分都实现了线程安全的功能，却不使用同步关键字(synchronized)。
 
常见阻塞队列：
　　BlockingQueue.class，阻塞队列接口
　　DelayQueue.class，阻塞队列，*队列，并且元素是Delay的子类，保证元素在达到一定时间后才可以取得到
public class T07_DelayQueue {

	// 执行定时任务

	static BlockingQueue<MyTask> tasks = new DelayQueue<>();

	static Random r = new Random();

	static class MyTask implements Delayed {

		long runningTime;

		MyTask(Long rt) {

			this.runningTime = rt;

		}

		@Override

		public int compareTo(Delayed o) {

			if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS))

				return -1;

			if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS))

				return 1;

			else

				return 0;

		}

		@Override

		public long getDelay(TimeUnit unit) {

			return unit.convert(runningTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);

		}

		@Override

		public String toString() {

			return "" + runningTime ;

		}

	}

	public static void main(String[] args) throws InterruptedException {

		long now = System.currentTimeMillis();

		MyTask t1 =new MyTask(now +1000);

		MyTask t2 =new MyTask(now +2000);

		MyTask t3 =new MyTask(now +1500);

		MyTask t4 =new MyTask(now +2500);

		MyTask t5 =new MyTask(now +500);

		tasks.put(t1);

		tasks.put(t2);

		tasks.put(t3);

		tasks.put(t4);

		tasks.put(t5);

		System.out.println(tasks);

		for(int i=0;i<5;i++) {

			System.out.println(tasks.take());

		}

	}

}

　　

　　BlockingDeque.class，双端阻塞队列接口
　　ArrayBlockingQueue.class，阻塞队列，数组实现。有界阻塞队列
public class T06_ArrayBlockingQueue {

	// 阻塞队列

	static BlockingQueue<String> strs = new ArrayBlockingQueue<>(10);

	static Random r =new Random();

	public static void main(String[] args) throws InterruptedException {

		for(int i=0;i<10;i++) {

			strs.put("a"+i);

		}

//		strs.add("aaa");//Exception in thread "main" java.lang.IllegalStateException: Queue full

//		strs.put("aaa"); // 满了会等待 阻塞

//		strs.offer("aaa");//不会报异常 有返回值  true or false

//		strs.offer("aaa",1,TimeUnit.SECONDS); // 按照时间段阻塞

	}

}

　　LinkedBlockingDeque.class，阻塞双端队列，链表实现
　　LinkedBlockingQueue.class，阻塞队列，链表实现，*，以下是该类对与生产者消费者模型的简单实现：

public class T05_LinkedBlockingQueue {

	// 阻塞队列

	static BlockingQueue<String> strs = new LinkedBlockingQueue<>();

	static Random r =new Random();

	//生产者消费者

	public static void main(String[] args) {

		new Thread(()->{

			for(int i=0;i<100;i++) {

				try {

					strs.put("a"+i);//如果满了就会等待阻塞

					System.err.println("a"+i);

					TimeUnit.MILLISECONDS.sleep(r.nextInt(1000));

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		},"p1").start();

		for(int i=0;i<5;i++) {

			new Thread(()->{

				while(true) {

					try {

						//take 如果空了 就会阻塞

						System.out.println(Thread.currentThread().getName()+" take - "+ strs.take());

					} catch (InterruptedException e) {

						e.printStackTrace();

					}

				}

			},"c" +i).start();

		}

	}

}

　　
　　SynchronousQueue.class，同步队列，但是队列长度为0，生产者放入队列的操作会被阻塞，直到消费者过来取，所以这个队列根本不需要空间存放元素；有点像一个独木桥，一次只能一人通过，还不能在桥上停留

public class T09_SynchronousQueue {

	// 特殊的 transferQueue ，队列为空的

	//任何添加东西都要直接丢给消费者的，不会往队列里加的

	public static void main(String[] args) throws InterruptedException {

		BlockingQueue<String> strs = new SynchronousQueue<>();

		new Thread(()->{

			try {

				System.out.println(strs.take());

			} catch (InterruptedException e) {

				e.printStackTrace();

			}

		}).start();

		strs.put("aaa");//阻塞，等待消费者消费

//		strs.add("aaa"); // 报错

		System.out.println(strs.size());

	}

}
 
常见非阻塞队列：
　　ConcurrentLinkedDeque.class，非阻塞双端队列，链表实现，*队列
　　ConcurrentLinkedQueue.class，非阻塞队列，链表实现
public class T04_ConcurrentQueue {

	public static void main(String[] args) {

		Queue<String> strs = new ConcurrentLinkedQueue<>();

		Queue<String> strs2 = new ConcurrentLinkedDeque<>();//双端队列

		for(int i=0;i<10;i++) {

			// 根据返回值来判断是否添加成功

			strs.offer("a"+i);

		}

		System.out.println(strs);//[a0, a1, a2, a3, a4, a5, a6, a7, a8, a9]

		System.out.println(strs.size());// 10

		System.out.println(strs.poll());// a0

		System.out.println(strs.size());// 9

		System.out.println(strs.peek());// a1

		System.out.println(strs.size());// 9

	}

}

　

转移队列：
　　TransferQueue.class，转移队列接口，生产者要等消费者消费的队列，生产者尝试把元素直接转移给消费者
　　LinkedTransferQueue.class，转移队列的链表实现，它比SynchronousQueue更快，transfer 方法有客户端准备消费，直接把消息直接传递给消费者，不放到队列里，没有消费者线程的话该线程会阻塞。但是可以调用 add put 王队列里丢，队列还是有容量的。
public class T08_TransferQueue {

	public static void main(String[] args) throws InterruptedException {

		LinkedTransferQueue<String> strs = new LinkedTransferQueue<>();

		new Thread(()->{

			try {

				System.out.println(strs.take());

			} catch (InterruptedException e) {

				e.printStackTrace();

			}

		}).start();

		strs.transfer("aaa");

//		new Thread(()->{

//			try {

//				System.out.println(strs.take());

//			} catch (InterruptedException e) {

//				e.printStackTrace();

//			}

//		}).start();

	}

}

　　

常见集合容器：
　　ConcurrentMap.class，并发Map的接口，定义了putIfAbsent(k,v)、remove(k,v)、replace(k,oldV,newV)、replace(k,v)这四个并发场景下特定的方法
　　ConcurrentHashMap.class，并发HashMap，ConcurrentHashMap在并发中效率比 HashTable高，因为 HashTable 在往里添加东西的时候药锁定整个对象，而 ConcurrentHashMap 分成了16段，插入的时候只锁定了其中的一段，其实就是把锁细粒度化了，因此在多线程情况下回比 hashTable高 ，同样也比 Collections.synchronizedMap(map1) 高。
　　ConcurrentSkipListMap.class，跳表数据结构，它也是NavigableMap的实现类（要求元素之间可以比较），只有你确实需要快速的遍历操作，并且可以承受额外的插入开销的时候，在高并发中要求排序才去使用它。
　　ConcurrentSkipListSet.class，和上面类似，只不过map变成了set
　　以下代码可以简单的比较一下几个容器再并发的情况下的效率问题
public class T01_ConcurrentMap {

	public static void main(String[] args) {

		Map<String,String> map =new ConcurrentHashMap<>();//并发性比较高

//		Map<String,String> map =new ConcurrentSkipListMap<>();//并发性高而且要求排序

//		Map<String,String> map  =new Hashtable<>();//并发性不是跟高

//		Map<String,String> map1 =new HashMap<>();

//		Map<String, String> map = Collections.synchronizedMap(map1);//并发性不是很高

		Random r =new Random();

		Thread[] ths  =new Thread[100];

		CountDownLatch latch =new CountDownLatch(ths.length);

		long start = System.currentTimeMillis();

		for(int i=0;i<ths.length;i++) {

			ths[i]=new Thread(()->{

				for(int j=0;j<10000;j++) {

					map.put("a"+r.nextInt(10000), "a"+r.nextInt(10000));

				}

				latch.countDown();

			});

		}

		Arrays.asList(ths).forEach(o->o.start());

		try {

			latch.await();

		} catch (InterruptedException e) {

			e.printStackTrace();

		}

		long end = System.currentTimeMillis();

		System.out.println(end-start);

	}

}
 
　　CopyOnWriteArrayList.class，copy-on-write（写时复制）模式的array list，每当需要插入元素，不在原list上操作，而是会新建立一个list，然后将原先的引用指向副本。适合读远远大于写的场景
　　CopyOnWriteArraySet.class，和上面类似，list变成set而已　
　　以下代码可以简单的比较一下几个容器再并发的情况下的效率问题　
　　 
public class T01_CopyOnWriteList {

	public static void main(String[] args) {

		List<String> lists =

//				new Vector<String>();//并发效率相对高

//				new ArrayList<String>();//这个并发会有问题

				new CopyOnWriteArrayList<String>();//效率比较低，读取非常快，读的时候不加锁

		Random r =new Random();

		Thread[] ths  =new Thread[100];

		for(int i=0;i<ths.length;i++) {

			ths[i]=new Thread(()->{

				for(int j=0;j<1000;j++) {

					lists.add("a"+r.nextInt(10000) );

				}

			});

		}

		long start = System.currentTimeMillis();

		Arrays.asList(ths).forEach(o->o.start());

		Arrays.asList(ths).forEach(o->{

			try {

				o.join();

			} catch (InterruptedException e) {

				e.printStackTrace();

			}

		});

		long end = System.currentTimeMillis();

		System.out.println(end-start);

		System.out.println(lists.size());

	}

}
上一篇：Selenium+Python自动化测试环境搭建和搭建过程遇到的问题解决


下一篇：ArcMap中地图输出（Options）选项显示不完整