JAVA并发工具类

一、分而治之

fork/join     二叉树  二分查找  快速排序  归并排序  mapreduce  动态规划

1、fork/join(工作密取)

  RecursiveTask要有返回值

  RecursiveAction没有返回值

  invoke(同步)

  submit(有返回结果异步)

  execute(没有返回结果异步)

2、countDownLatch(闭锁)只能使用一次

作用:直指一个线程等待其他的线程执行完后再执行,(就是一个加强版的join())

用一个线程可以有多个扣除点,扣除点可以在程序的中间,await可以有多个。

3、CyclicBarrier可以使用一次

作用:一组工作线程之间需要进行协作

4、Semaphore(控制同时访问某个特定资源的线程数量,例如用于流量控制)

Semaphore的大坑就是:release方法可以无限增加;所以一般会定义两个Semaphore变量,结合来使用。

Semaphore只是控制流量,获取许可证;需要和锁配合使用才能完成总的并发控制

5、Exchanger用于两个线程间的数据交换

6、Future

  isDone()不管正常或者异常结束又或者自己取消,都会返回true

  isCancelled如果任务在完成前被取消,返回true;其他情况返回false

  cancel()(1)如果任务已经完成,或者已经取消或者由于某些原因不能取消,则返回false;

       (2)如果任务还没有执行,则会返回true,并且异步任务不会执行

     (3)如果任务已经开始,但是还没有执行完成,则返回true(mayInterrupIfRunning为true时,会中断线程)

FutureTask  treiber  AQS实现

二、原子类(CAS原理)乐观锁的思想

悲观锁容易发生死锁

原子操作实现方式:1、synchronized;2、原子类

1、cas的原理:

cpu硬件指令集的支持

包括三个运算参数:内存地址V,期望值A,新值B

不能更新成功的话,会一直自旋

2、cas的问题

ABA问题

开销问题

只能保证一个共享变量的原子操作

3、AtomicBoolean,AtomicInteger,AtomicLong

4、AtomicReference(多个变量同时进行更新,把多个变量包装成引用类),AtomicStampedReference()(有具体的版本号),AtomicMarkableReference(boolean标志版本)(只关心有没有变化)

5、AtomicIntegerArray,AtomicLongArray,AtomicReferenceArray[原子数组类实质上又另外创建了一个和原数组完全独立,但一模一样的数组]

6、源自更新字段类AtomicIntegerFieldUpdate,AtomicLongFieldUpdate,AtomicReferenceFieldUpdate

原子操作底层用的是CAS操作

三、显示锁

1、ReentrantLock(优先使用synchronized)

排它锁

线程退出时(执行完时),它所拥有的资源都会被释放

Synchronized进入调锁的过程中后:Synchronized不提供中断和超时机制

Synchronized消耗更少一点,Synchronized是语言特性,而显示锁是一种类

Synchronized方法块内抛异常,代码脱离方法快,锁会自动释放

Lock获取锁可以被中断,超时获取锁,尝试获取锁

非公平锁充分的利用了线程被唤醒的这段时间

公平锁(排队)与非公平锁(可以插队,效率更高)

synchronized尽量使用notifyAll;ReentrantLock尽量使用signal

一个Lock对应一个一个Condition

一个Lock对应多个Condition

lockInterruptibly

tryLock

tryLock(timeout)

2、读写锁ReadWriteLock

读是共享的,写是排它锁,最适合于读多写少的场景

四、LockSupport(基础工具)

五、AQS(抽象队列同步器)

1final void lock() { 2 if (compareAndSetState(0, 1)) 3 setExclusiveOwnerThread(Thread.currentThread()); 4 else 5 acquire(1); 6 }

这个方法是ReentrantLock.NonfairSync.lock()方法

1public final void acquire(int arg) { 2 if (!tryAcquire(arg) && 3 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 4 selfInterrupt(); 5 }

这个方法就是尝试获取独占所同步状态,这是AQS的方法

1protected final boolean tryAcquire(int acquires) { 2 return nonfairTryAcquire(acquires); 3 }

 这个方法是由ReentrantLock.NonfairSync.tryAcquire实现

1final boolean nonfairTryAcquire(int acquires) { 2 final Thread current = Thread.currentThread(); 3 int c = getState(); 4 if (c == 0) { 5 if (compareAndSetState(0, acquires)) { 6 setExclusiveOwnerThread(current); 7 return true; 8 } 9 } 10 else if (current == getExclusiveOwnerThread()) { 11 int nextc = c + acquires; 12 if (nextc < 0) // overflow 13 throw new Error("Maximum lock count exceeded"); 14 setState(nextc); 15 return true; 16 } 17 return false; 18 }

这个方法是由ReentrantLock.Sync.nonfairTryAcquire实现(尝试获取独占锁,或者重入独占锁)

1private Node addWaiter(Node mode) { 2 Node node = new Node(Thread.currentThread(), mode); 3 // Try the fast path of enq; backup to full enq on failure 4 Node pred = tail; 5 if (pred != null) { 6 node.prev = pred; 7 if (compareAndSetTail(pred, node)) { 8 pred.next = node; 9 return node; 10 } 11 } 12 enq(node); 13 return node; 14 }

这个方法是AQS的方法,这个方法是(1)先设置尾节点compareAndSetTail(2)如果前面设置没成功,就进入死循环enq方法,这个方法里会判断为节点是否为空,如果为空,就先设置head指向New Node()空节点,然后再循环;如果不为空,就设置compareAndSetTail,如果设置成功就可以退出了,否则继续循环。

1private Node enq(final Node node) { 2 for (;;) { 3 Node t = tail; 4 if (t == null) { // Must initialize 5 if (compareAndSetHead(new Node())) 6 tail = head; 7 } else { 8 node.prev = t; 9 if (compareAndSetTail(t, node)) { 10 t.next = node; 11 return t; 12 } 13 } 14 } 15 }

这个就是AQS的enq方法

1final boolean acquireQueued(final Node node, int arg) { 2 boolean failed = true; 3 try { 4 boolean interrupted = false; 5 for (;;) { 6 final Node p = node.predecessor(); 7 if (p == head && tryAcquire(arg)) { 8 setHead(node); 9 p.next = null; // help GC 10 failed = false; 11 return interrupted; 12 } 13 if (shouldParkAfterFailedAcquire(p, node) && 14 parkAndCheckInterrupt()) 15 interrupted = true; 16 } 17 } finally { 18 if (failed) 19 cancelAcquire(node); 20 } 21 }

 这个方法就是加入到AQS队列以后发生的等待操作

Springtemplate就是模板设计模式

state有三个方法(getState,setState,compareAndSetState)

state为0时,可以直接获取锁,不需要加入AQS队列

state不为0时,需要尝试获取锁

模板方法设计模式

重入锁:

读写锁:

分布式锁:抢到本地锁的线程再去抢分布式锁;redis性能好;zookeeper稳定性好。

读写锁的实现

读锁的实现

1 public final void acquireShared(int arg) { 2 if (tryAcquireShared(arg) < 0) 3 doAcquireShared(arg); 4 }

AQS的模板方法:

1protected final int tryAcquireShared(int unused) { 2 Thread current = Thread.currentThread(); 3 int c = getState(); 4 if (exclusiveCount(c) != 0 && 5 getExclusiveOwnerThread() != current) 6 return -1; 7 int r = sharedCount(c); 8 if (!readerShouldBlock() && 9 r < MAX_COUNT && 10 compareAndSetState(c, c + SHARED_UNIT)) { 11 if (r == 0) { 12 firstReader = current; 13 firstReaderHoldCount = 1; 14 } else if (firstReader == current) { 15 firstReaderHoldCount++; 16 } else { 17 HoldCounter rh = cachedHoldCounter; 18 if (rh == null || rh.tid != getThreadId(current)) 19 cachedHoldCounter = rh = readHolds.get(); 20 else if (rh.count == 0) 21 readHolds.set(rh); 22 rh.count++; 23 } 24 return 1; 25 } 26 return fullTryAcquireShared(current); 27 }

尝试获取读锁:

  1、如果当前线程是写线程,直接返回-1(加入等待队列)

  2、如果当前头结点的后继节点是读线程直接返回1获取到所

  3、如果当前头结点的后继节点是写线程,直接执行fullTryAcquireShared方法(这个策略完全是为了照顾写线程,或者说是防止写线程过分的饥渴)

1final boolean readerShouldBlock() { 2 return apparentlyFirstQueuedIsExclusive(); 3 }

as

1final boolean apparentlyFirstQueuedIsExclusive() { 2 Node h, s; 3 return (h = head) != null && 4 (s = h.next) != null && 5 !s.isShared() && 6 s.thread != null; 7 }

as

1final int fullTryAcquireShared(Thread current) { 2 /* 3 * This code is in part redundant with that in 4 * tryAcquireShared but is simpler overall by not 5 * complicating tryAcquireShared with interactions between 6 * retries and lazily reading hold counts. 7 */ 8 HoldCounter rh = null; 9 for (;;) { 10 int c = getState(); 11 if (exclusiveCount(c) != 0) { 12 if (getExclusiveOwnerThread() != current) 13 return -1; 14 // else we hold the exclusive lock; blocking here 15 // would cause deadlock. 16 } else if (readerShouldBlock()) { 17 // Make sure we're not acquiring read lock reentrantly 18 if (firstReader == current) { 19 // assert firstReaderHoldCount > 0; 20 } else { 21 if (rh == null) { 22 rh = cachedHoldCounter; 23 if (rh == null || rh.tid != getThreadId(current)) { 24 rh = readHolds.get(); 25 if (rh.count == 0) 26 readHolds.remove(); 27 } 28 } 29 if (rh.count == 0) 30 return -1; 31 } 32 } 33 if (sharedCount(c) == MAX_COUNT) 34 throw new Error("Maximum lock count exceeded"); 35 if (compareAndSetState(c, c + SHARED_UNIT)) { 36 if (sharedCount(c) == 0) { 37 firstReader = current; 38 firstReaderHoldCount = 1; 39 } else if (firstReader == current) { 40 firstReaderHoldCount++; 41 } else { 42 if (rh == null) 43 rh = cachedHoldCounter; 44 if (rh == null || rh.tid != getThreadId(current)) 45 rh = readHolds.get(); 46 else if (rh.count == 0) 47 readHolds.set(rh); 48 rh.count++; 49 cachedHoldCounter = rh; // cache for release 50 } 51 return 1; 52 } 53 } 54 }

循环尝试获取读锁:

  1、如果当前线程是写线程,直接返回-1(加入等待队列)

  2、如果当前头结点的后继节点是写线程且不是锁的重入状态,那么就直接退出-1(加入等待队列)

  3、如果写线程超量,直接抛出异常

  4、如果修改状态成功,直接返回1,获取读锁成功

1private void doAcquireShared(int arg) { 2 final Node node = addWaiter(Node.SHARED); 3 boolean failed = true; 4 try { 5 boolean interrupted = false; 6 for (;;) { 7 final Node p = node.predecessor(); 8 if (p == head) { 9 int r = tryAcquireShared(arg); 10 if (r >= 0) { 11 setHeadAndPropagate(node, r); 12 p.next = null; // help GC 13 if (interrupted) 14 selfInterrupt(); 15 failed = false; 16 return; 17 } 18 } 19 if (shouldParkAfterFailedAcquire(p, node) && 20 parkAndCheckInterrupt()) 21 interrupted = true; 22 } 23 } finally { 24 if (failed) 25 cancelAcquire(node); 26 } 27 }

 写锁的实现

1public final void acquire(int arg) { 2 if (!tryAcquire(arg) && 3 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 4 selfInterrupt(); 5 }

as

1protected final boolean tryAcquire(int acquires) { 2 Thread current = Thread.currentThread(); 3 int c = getState(); 4 int w = exclusiveCount(c); 5 if (c != 0) { 6 // (Note: if c != 0 and w == 0 then shared count != 0) 7 if (w == 0 || current != getExclusiveOwnerThread()) 8 return false; 9 if (w + exclusiveCount(acquires) > MAX_COUNT) 10 throw new Error("Maximum lock count exceeded"); 11 // Reentrant acquire 12 setState(c + acquires); 13 return true; 14 } 15 if (writerShouldBlock() || 16 !compareAndSetState(c, c + acquires)) 17 return false; 18 setExclusiveOwnerThread(current); 19 return true; 20 }

1、如果当前锁是共享锁,那么直接返回false

2、如果可重入写线程数量太多超过最大值,抛出异常

3、可重入线程正常直接返回true

4、如果当前没有锁,且自己抢到了锁,直接返回true

5、如果当前没有锁,且没有抢到锁,直接返回false

1private Node addWaiter(Node mode) { 2 Node node = new Node(Thread.currentThread(), mode); 3 // Try the fast path of enq; backup to full enq on failure 4 Node pred = tail; 5 if (pred != null) { 6 node.prev = pred; 7 if (compareAndSetTail(pred, node)) { 8 pred.next = node; 9 return node; 10 } 11 } 12 enq(node); 13 return node; 14 }

as

1private Node enq(final Node node) { 2 for (;;) { 3 Node t = tail; 4 if (t == null) { // Must initialize 5 if (compareAndSetHead(new Node())) 6 tail = head; 7 } else { 8 node.prev = t; 9 if (compareAndSetTail(t, node)) { 10 t.next = node; 11 return t; 12 } 13 } 14 } 15 }

as

1final boolean acquireQueued(final Node node, int arg) { 2 boolean failed = true; 3 try { 4 boolean interrupted = false; 5 for (;;) { 6 final Node p = node.predecessor(); 7 if (p == head && tryAcquire(arg)) { 8 setHead(node); 9 p.next = null; // help GC 10 failed = false; 11 return interrupted; 12 } 13 if (shouldParkAfterFailedAcquire(p, node) && 14 parkAndCheckInterrupt()) 15 interrupted = true; 16 } 17 } finally { 18 if (failed) 19 cancelAcquire(node); 20 } 21 }

五、并发容器

位运算的关键【a%2^n==a&(2^n-1)】

hashmap会造成死循环:扩容时头插法惹的祸

1、ConcurrentHashMap

get   put   putIfAbsent

putIfAbsent返回原来那个key所对应的值

 取模操作相当于位与操作

位运算用途:权限控制和商品属性

jdk1.7实现concurrenthashmap:segment数组(16缺省并发度这个数必须是2^n,这个数组也不允许扩容,使用hash值的高位部分进行hash)(ReentrantLock)创建segment元素时使用CAS保证线程安全  每一个segment下面包含一个hashentry数组table[]这个数组大小也必须是2^n,通过整个hash值进行hash,负载因子是0.75;①定位segment②定位hashentry③循环链表;get方法不加锁,只用了volatile{final hash;final key;volatile value;volatile next}      put方法 拿锁tryLock() 和lock()结合使用,提前new Node(key,value)   扩容也最好是2的倍数,迁移的时候会变更迁移方便,扩容会扩与扩原来的2倍   

这两个方法尽量少使用:(size方法首先算两遍然后可能会全部加锁   containsValue也需要全部加锁)

存在的问题:弱一致性,获取的过程中整个结构可能发生调整,因为get方法是没有加锁的

扩容:stride步长+forwarding

再设置sizeCtl

size()=basecount+countercells

强一致性:Collections.synchronizedMap(m),HashTable

jdk1.8实现concurrenthashmap:CAS+synchronized

数组+链表+红黑树(6~8)

元素是Node(final,final,volatile,volatile)不是Entry(TreeNode继承于node(树叶),TreeBin(树根),Node);treebin之中使用了读写锁;

forwardingnode扩容的时候需要用到占位

sizeCtl(-1正在初始化)(0是默认值)(-n表示有多少线程正在扩容)(正数相当于负载因子)

tabAt,setTabAt,casTabAt

构造函数什么也没做

get方法:数组+红黑树+链表

put方法:初始化table,只有一个线程能够初始化成功(sizectl开控制这个并发);CAS操作把sizectl置为-1的线程初始化table;sizectl设置为阈值;CAS操作向数组里放入值;正在扩容时,帮助扩容;synchronized(锁住树根)网链表或者红黑树里面插入值【尾插法】,树与链表的转换;扩容

负载因子仍然是0.75

sizeCtl

Node

get put

jkd1.8实现hashmap:treenode(继承于linkedhashmap.entry)

2、跳表

增加链表的快速查找性(redis,lucence)

3、ConcurrentLinkedQueue

无界非租塞队列,线程安全

4、CopyOnWriteArrayList

CopyOnWriteArraySet

最适应读多写少的并发场景(白名单,黑名单)

内存占用严重,只能保证最终一致性

批量提交

5、阻塞队列(尽量使用有界队列)(等待通知模式)

有界就是put会阻塞的队列

无界就是put不会阻塞的队列

生产者消费者模式

ArrayBlockingQueue(有界阻塞队列,先进先出原则,必须设定初始大小,只用一个锁,直接插入元素)【生产者和消费者是同一把锁】

LinkedBlockingQueue(有界阻塞队列,先进先出原则,可以不设定初始大小默认就是Integer.MAX_Value,用了两个锁,插入元素时需要转换)【生产者和消费者是两把锁】

PriorityBlockingQueue(无界阻塞队列)(默认,按照自然顺序,要么实现compareTo()方法,指定构造参数Comparator)

DelayQueue(无界阻塞队列)(实现自己的缓冲系统,订单到期,限时支付)(延时获取元素)消息中间件

SynchronousQueue(不存储元素)(每一个put操作都要等待一个take操作)(Exchanger)

LinkedTransferQueue(无界阻塞队列)相比LinkedBlockingQueue,多了两个方法transfer()必须要消费者消费了以后方法才会返回(先直接给消费者,消费者没有则放入队列阻塞);tryTransfer()无论消费者是否接收,方法都立即返回(先给消费者,消费者没有就直接返回)

LinkedBlockingDeque(双向阻塞队列(fork/join工作密取))可以从头和尾插入和删除元素

六、线程池

如果开启prestartAllCoreThreads,那么提交任务就会把线程直接放到queue里面;当queue里面满了的时候,就直接执行线程数直到线程最大值;如果队列设置太大;那么最大线程数和拒绝策略就没什么意义了。;队列的大小最好大于核心线程数,但是不能过大。

1、ThreadPoolExecutor

  keepalivetime不对coresthread起作用,请对其他线程起作用

  ThreadFactory(作用是创建线程名字)

  RejectedExecutionHandler(拒绝策略)

    AbortPolicy(直接抛出异常,默认策略)

    CallerRunsPolicy让提交的线程执行该线程

prestartAllCoreThreads设置一开始就会起来corepool线程

allowCoreThreadTimeOut是否允许核心线程数也超时退出

    DiscardOldestPolicy(丢弃队列里面最老的任务)

    DiscardPolicy直接抛弃

线程池的AOP模式

计算密集型:加密,大数分解,正则【线程数适当小一点,机器的cpu核心数+1,操作系统调用线程就会造成页缺失,+1为了防止也确实】

IO密集型:读取文件,数据库连接,网络通讯,rpc【线程数适当大一点:cpu的核心数×2】(NCPU*UCPU*(1+W/C))

混合型:尽量拆分成为IO密集型和计算密集型(如果使用的时间相差过大,就不需要拆分了)

FixedThreadPool(创建固定数量的线程池,适用于负载较重的服务器,使用了无界队列)

CachedThreadPool会根据需要创建新线程(执行很多短期异步任务的程序)使用了SynchronousQueue阻塞队列

SingleThreadPool创建单个线程,需要顺序保证执行任务,没有线程安全问题,使用了无界队列(保证任务串行执行)

WorkStealingPool工作密取(fork/join)

2、ScheduledThreadPoolExecutor

  (1)newSingleThreadScheduledExecutor只包含一个线程,只需要单个线程执行周期任务,保证顺序的执行各个任务

  (2)newScheduledThreadPool可以包含多个线程的,线程执行周期任务,适度控制后台线程数量的时候

  方法说明:

    schedule只执行一次,可以延时执行(抛出异常以后,任务直接中止执行;需要用try---catch)

    scheduleAtFixedRate任务超时怎么办?下一个任务马上开始执行(抛出异常以后,任务直接中止执行;需要用try---catch)

    scheduleWithFixedDelay(抛出异常以后,任务直接中止执行;需要用try---catch)

3、CompletionService

4、CompleteableFuture

七、线程安全性

栈封闭

无状态(无状态的类即没有任何成员变量的类)【servlet是线程不安全的】

让类不可变(final,String,包装类)只要不变,所有的成员变量都加上final关键字(Akka)

volatile保证类的可见性(最适合一个线程写,多个线程读的情景)(1)可见性(2)禁止重排序

加锁和CAS

安全的发布

ThreadLocal

Collections.synchronizedList()和使用synchronized包装普通类(继承和委托)

AtomicReference,CopyOnWriteArrayList

死锁是由于加锁造成的(争夺资源顺序,操作者>资源数)(数据库事务也可能产生死锁)jps -v    ;jstack 7412

解决办法:强制拿锁顺序必须一致,动态死锁原生hashcode解决(拿第三次锁);tryLock尝试加两个锁(但为了避免活锁,也要等待随机时间);

活锁

线程饥饿:读写锁的机制

性能:线程上下文切换,锁同步,页缺失

vmstat

内存屏障指令,对性能有影响

阻塞:将导致线程的挂起

解决办法:

同时锁两个锁

减少锁的范围

减少锁的力度(通过增加锁的数量)

锁分段

替换独占锁(读写锁的应用,CAS替换synchronized,并发容器)

避免多余的锁(就是把两把锁合成一把锁,锁的粗化)

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

皕杰报表之UUID

​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段,为你的工作保驾护航

有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )