Java高并发编程-并发容器
学习马士兵老师的公开课，整理的笔记
题目：有N张火车票，每张票都有一个编号，同时有10个窗口对外售票
实现一：使用ArrayList
因为list的remove操作是非原子性的，所以多个线程在同时remove的时候可能会操作同一张票
代码如下：
public class TicketSeller1 {
private static List<String> tickets = new ArrayList<>();
static{
for (int i = 0; i < 10000; i++) {
tickets.add("票编号：" + i);
}
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
while(tickets.size() > 0){
System.out.println("销售了>>>>>" + tickets.remove(0));
}
}).start();
}
}
}
输出如下，发生了重售：
销售了>>>>>票编号：7835
销售了>>>>>票编号：7792
销售了>>>>>票编号：7565
销售了>>>>>票编号：9999
销售了>>>>>票编号：9542
销售了>>>>>票编号：9515
销售了>>>>>票编号：9242
实现二：使用Vector或者Collections.synchronizedXXX
虽然remove方法是原子性的但是判断与remove的操作合在一起就不是原子性的所以仍然会出现问题
代码如下：
public class TicketSeller2 {
private static Vector<String> tickets = new Vector<String>();
static {
for (int i = 0; i < 10000; i++) {
tickets.add("票编号：" + i);
}
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
while(tickets.size() > 0){
try {
TimeUnit.MILLISECONDS.sleep(10);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("销售了>>>" + tickets.remove(0));
}
}).start();
}
}
}
运行结果如下
销售了>>>票编号：9993
销售了>>>票编号：9992
Exception in thread "Thread-9" java.lang.ArrayIndexOutOfBoundsException: Array index out of range: 0
at java.util.Vector.remove(Vector.java:831)
at thread.demo_024.TicketSeller2.lambda$main$0(TicketSeller2.java:37)
at java.lang.Thread.run(Thread.java:748)
销售了>>>票编号：9995
实现三：使用同步代码块锁住容器-保证正确性但效率低
就算操作A和B都是同步的，但A和B组成的复合操作也未必是同步的，仍然需要自己进行同步 就像这个程序，判断size和进行remove必须是一整个的原子操作
代码如下：
public class TicketSeller3 {
private static List<String> tickets = new LinkedList<String>();
static {
for (int i = 0; i < 10000; i++) {
tickets.add("票编号：" + i);
}
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
while(true){
synchronized(tickets){
if(tickets.size() <= 0) break;
try {
TimeUnit.MILLISECONDS.sleep(10);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("销售了>>>" + tickets.remove(0));
}
}
}).start();
}
}
}
实现四：使用ConcurrentQueue提高并发性
使用JDK1.5之后提供的并发队列ConcurrentLinkedQueue存储元素,其底层使用CAS实现而非加锁实现的,其效率较高。
并发队列ConcurrentLinkedQueue的poll()方法会尝试从队列头中取出一个元素,若获取不到,则返回null,对其返回值做判断可以实现先取票后判断,可以避免加锁。
代码如下：
public class TicketSeller4 {
private static Queue<String> tickets = new ConcurrentLinkedQueue<String>();
static {
for (int i = 0; i < 10000; i++) {
tickets.add("票编号：" + i);
}
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
while(true){
String s = tickets.poll();
if(s == null) break;
try {
TimeUnit.MILLISECONDS.sleep(10);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("销售了>>>>" + s);
}
}).start();
}
}
}
2. 并发容器
Map/Set 它们比较类似 set没有value只有key而已
（1）非并发容器
HashMap,TreeMap,LinkedHashMap
（2）并发容器
HashTable,Collections.sychronizedXXX,ConcurrentMap
- 并发量不大的时候使用
HashTable,Collections.sychronizedXXX他们同步实现的原理类似都是直接给整个容器加上锁，所以效率会比较差
Collections.sychronizedXXX 的用法是使用装饰器模式,调用其构造方法并传入一个XXX的实现类,返回一个同步的XXX容器
代码如下：
public class T03_SynchronizedMap {
public static void main(String[] args) {
HashMap<String> hashMap = new HashMap<String>();
Map<String> synchronizedMap = Collections.synchronizedMap(hashMap); //此时的集合是同步的
}
}
- 并发量比较高的时候使用
ConcurrentMap，有两个实现类：ConcurrentHashMap: 使用哈希表实现,key是无序的ConcurrentSkipListMap: 使用跳表实现,key是有序的
其同步的实现原理在JDK1.8前后不同
- 在JDK1.8以前,其实现同步使用的是分段锁,将整个容器分为16段(Segment),每次操作只锁住操作的那一段,是一种细粒度更高的锁.
- 在JDK1.8及以后,其实现同步用的是Node+CAS.
public class T01_ConcurrentMap {
public static void main(String[] args) {
Map<String, String> map = new ConcurrentHashMap<String, String>();
//Map<String, String> map = new ConcurrentSkipListMap<String, String>(); //高并发并且排序
//Map<String, String> map = new Hashtable<>();
//Map<String, String> map = new HashMap<String, String>();
Random random = new Random();
Thread[] threads = new Thread[100];
CountDownLatch latch = new CountDownLatch(threads.length);
long start = System.currentTimeMillis();
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(()->{
for(int j=0; j<10000;j++) map.put("a" + random.nextInt(100000), "a" + random.nextInt(100000));
latch.countDown();
});
}
Arrays.asList(threads).forEach(t->t.start());
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
System.out.println(end-start);
}
}
Queue
低并发队列
适用于并发量比较低的队列有:Vector和SynchronizedList,其中vector类似HashTable,是JDK1.2就提供的类;SynchronizedList类似SynchronizedMap使用装饰器模式,其构造函数接受一个List实现类并返回同步List,在java.util.Collections包下.
它们实现同步的原理都是将所有方法用同步代码块包裹起来.
写时复制CopyOnWriteList
CopyOnWriteArrayList位于java.util.concurrent包下,它实现同步的方式是: 当发生写操作(添加,删除,修改)时,就会复制原有容器然后对新复制出的容器进行写操作,操作完成后将引用指向新的容器.其写效率非常低,读效率非常高
- 为什么读操作不需要加锁？因为本身
CopyOnWriteList就不存在脏读的情况
优点: 读写分离,使得读操作不需要加锁,效率极高
缺点: 写操作效率极低
应用场合: 应用在读少写多的情况,如事件***
高并发队列
| 方法 | 抛出异常 | 返回特殊值 | 一直阻塞 (非阻塞队列不可用) | 阻塞一段时间 (非阻塞队列不可用) |
|---|---|---|---|---|
| 插入元素 | add(element) | offer(element) | put(element) | offer(element,time,unit) |
| 移除首个元素 | remove() | poll() | take() | poll(time,unit) |
| 返回首个元素 | element() | peek() |
对于高并发队列,若使用不同的方法对空队列执行查询和删除,以及对满队列执行插入,会产生不同行为
- 抛出异常: 使用
add(),remove(),element()方法,若执行错误操作会直接抛出异常 - 返回特殊值: 若使用
offer(),poll(),peek()方法执行错误操作会返回false或null,并放弃当前错误操作,不抛出异常. - 一直阻塞: 若使用
put(),take()方法执行错误操作,当前线程会一直阻塞直到条件允许才唤醒线程执行操作. - 阻塞一段时间: 若使用
offer(),poll()方法并传入时间单位,会将当前方法阻塞一段时间,若阻塞时间结束后仍不满足条件则返回false或null,并放弃当前错误操作,不抛出异常.
非阻塞队列ConcurrentLinkedQueue
非阻塞队列使用CAS保证操作的原子性,不会因为加锁而阻塞线程.类似于ConcurrentMap
阻塞队列BlockingQueue
阻塞队列的常用实现类有LinkedBlockingQueue,ArrayBlockingQueue,DelayedQueue,TransferQueue,SynchronousQueue. 分别对应于不同的应用场景.
延迟队列DelayedQueue
它是一种无界队列，延迟队列DelayedQueue中存储的元素必须实现Delay接口,其中定义了getDelay()方法;而Delay接口继承自Comparable接口,其中定义了compareTo()方法.各方法作用如下:
getDelay(): 规定当前元素的延时,Delay类型的元素必须要等到其延时过期后才能从容器中取出,提前取会取不到.compareTo():规定元素在容器中的排列顺序,按照compareTo()的结果升序排列.
public class T08_TransferQueue {
public static void main(String[] args) throws InterruptedException {
LinkedTransferQueue<String> strings = new LinkedTransferQueue<String>();
/*new Thread(()->{
try {
System.out.println(strings.take());
} catch (Exception e) {
e.printStackTrace();
}
}).start();*/
//strings.transfer("aaa");
strings.put("aaa");
new Thread(()->{
try {
System.out.println(strings.take());
} catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}
执行结果：等待时间长的先往外拿走
[1570266175973, 1570266177973, 1570266177473, 1570266176473, 1570266176973]
1570266175973
1570266176973
1570266176473
1570266177473
阻塞消费队列TransferQueue
TransferQueue继承自BlockingQueue,向其中添加元素的方法除了BlockingQueue的add(),offer(),put()之外,还有一个transfer()方法,该方法会使当前线程阻塞直到消费者将该线程消费为止.
DelayedQueue的应用场景：服务器端的消息队列
transfer()与put()的区别: put()方法会阻塞直到元素成功添加进队列,transfer()方法会阻塞直到元素成功被消费.
TransferQueue特有的方法如下:
-
transfer(E):阻塞当前线程直到元素E成功被消费者消费 -
tryTransfer(E):尝试将当前元素送给消费者线程消费,若没有消费者接受则返回false且放弃元素E,不将其放入容器中. -
tryTransfer(E,long,TimeUnit):阻塞一段时间等待消费者线程消费,超时则返回false且放弃元素E,不将其放入容器中. -
hasWaitingConsumer():指示是否有阻塞在当前容器上的消费者线程 -
getWaitingConsumerCount():返回阻塞在当前容器上的消费者线程的个数
public class T08_TransferQueue {
public static void main(String[] args) throws InterruptedException {
LinkedTransferQueue<String> strings = new LinkedTransferQueue<String>();
/*new Thread(()->{
try {
System.out.println(strings.take());
} catch (Exception e) {
e.printStackTrace();
}
}).start();*/
//这种情况就会一直等待消费者的出现而不执行下面的代码
strings.transfer("aaa");
//strings.put("aaa");
new Thread(()->{
try {
System.out.println(strings.take());
} catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}
零容量的阻塞消费队列SynchronousQueue
SynchronousQueue是一种特殊的TransferQueue,特殊之处在于其容量为0. 因此对其调用add(),offer()方法都会使程序发生错误(抛出异常或阻塞线程).只能对其调用put()方法,其内部调用transfer()方法,将元素直接交给消费者而不存储在容器中。
public class T09_Synchronized {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> strings = new SynchronousQueue<String>();
new Thread(()->{
try {
System.out.println(strings.take());
} catch (Exception e) {
e.printStackTrace();
}
}).start();
strings.put("aaa"); //阻塞等待消费者消费
//strings.add("aaa");
System.out.println(strings.size());
}
}
