ReentrantLock
可重入
可中断
可限时
公平锁
简单示例
class ReentrantLockTest implements Runnable{
private static ReentrantLock reentrantLock = new ReentrantLock();
private static int i=0;
@Override
public void run() {
for (int j=0;j<10000;j++){
reentrantLock.lock();
try {
i++;
}finally {
reentrantLock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
Thread thread1 = new Thread(reentrantLockTest);
Thread thread2 = new Thread(reentrantLockTest);
thread1.start();
thread2.start();
thread1.join();
thread2.join();
System.out.println(i);
}
}
可重入
可重入锁的基本原理：锁也是一个类。
类的内部lock()方法，当线程第一个需要获取锁的时候，将当前线程保存到类中，并且将锁的状态设置为false，计数器+1，下次当某一个线程来获取锁的时候，lock()方法while循环判断，如果是不是当前线程并且锁的状态是false，就等待（并且一直尝试获取锁），如果是当前线程或者锁的状态为true，就继续加锁，计数器+1；
类的内部unlock()方法，首先判断是不是当前线程调用unlock方法（不是抛出异常），调用成功就将计数器-1，如果计数器=0，就将锁设置为true，此时其他的线程可以来获取锁了，否则只有当前线程才可以获取锁。
class ReentrantLockTest implements Runnable{
private static ReentrantLock reentrantLock = new ReentrantLock();
private static int i=0;
@Override
public void run() {
for (int j=0;j<10000;j++){
reentrantLock.lock();
reentrantLock.lock();
try {
i++;
}finally {
reentrantLock.unlock();
reentrantLock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
Thread thread1 = new Thread(reentrantLockTest);
Thread thread2 = new Thread(reentrantLockTest);
thread1.start();
thread2.start();
thread1.join();
thread2.join();
System.out.println(i);
}
}
可中断
当一个线程处于死锁或者长期等待状态，可以将该线程强制中断；
class DeadLockChecker{
private final static ThreadMXBean mxBean= ManagementFactory.getThreadMXBean();
private static Runnable runnable = new Runnable(){
@Override
public void run() {
while (true){
long[] deadlockedThreads = mxBean.findDeadlockedThreads();
if (deadlockedThreads!=null){
ThreadInfo[] threadInfo = mxBean.getThreadInfo(deadlockedThreads);
for (Thread t:Thread.getAllStackTraces().keySet()){
for (int i=0;i<threadInfo.length;i++){
if (t.getId()==threadInfo[i].getThreadId()){
t.interrupt();
}
}
}
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
public static void check(){
Thread thread = new Thread(runnable);
//设置守护线程;
thread.setDaemon(true);
thread.start();
}
}
class ReentrantLockTest implements Runnable{
private static ReentrantLock reentrantLock1 = new ReentrantLock();
private static ReentrantLock reentrantLock2 = new ReentrantLock();
private static int i=0;
private int x;
public ReentrantLockTest(int x) {
this.x = x;
}
@Override
public void run() {
//为了构造死锁
try {
if (x==1){
//和reentrantLock1.lock()差不多,都是加锁，但是lockInterruptibly()可以响应中断
reentrantLock1.lockInterruptibly();
Thread.sleep(500);
reentrantLock2.lockInterruptibly();
}
else {
reentrantLock2.lockInterruptibly();
Thread.sleep(500);
reentrantLock1.lockInterruptibly();
}
} catch (InterruptedException e) {
//可以处理其他的事情
e.printStackTrace();
}finally {
if (reentrantLock1.isHeldByCurrentThread()){
reentrantLock1.unlock();
}
if (reentrantLock2.isHeldByCurrentThread()){
reentrantLock2.unlock();
}
System.out.println("线程"+Thread.currentThread().getId()+"退出");
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest(1);
ReentrantLockTest reentrantLockTest2 = new ReentrantLockTest(2);
Thread thread1 = new Thread(reentrantLockTest);
Thread thread2 = new Thread(reentrantLockTest2);
thread1.start();
thread2.start();
//检查死锁,中断死锁（可以放在thread1和thread2前面,任何位置）
DeadLockChecker.check();
thread1.join();
thread2.join();
System.out.println(i);
}
}
可限时
class ReentrantLockTest implements Runnable{
private static ReentrantLock reentrantLock = new ReentrantLock();
@Override
public void run() {
try {
//表示当某一个线程过来获取锁,最多等待2秒,如果这个锁还没有被释放,直接跳过,不需要在等待,可以执行其他的事情了
if (reentrantLock.tryLock(2, TimeUnit.SECONDS)){
System.out.println("线程 "+Thread.currentThread().getId()+" get lock succeed");
Thread.sleep(3000);
}
else {
System.out.println("线程 "+Thread.currentThread().getId()+" get lock filed");
}
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
//会执行两次xx,一次yy,所以我们需要判断reentrantLock是否是当前线程所有,
//否则会抛出异常,原因在于有一个线程没有获取锁所以释放不了,就抛出异常;
System.out.println("xx");
if (reentrantLock.isHeldByCurrentThread()){
System.out.println("yy");
reentrantLock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
Thread thread1 = new Thread(reentrantLockTest);
Thread thread2 = new Thread(reentrantLockTest);
thread1.start();
thread2.start();
thread1.join();
thread2.join();
System.out.println("end");
}
}
公平锁
性能较非公平锁差很多，它要处理排队的问题，如果没有特殊的需要，不考虑公平锁
一般情况下，先申请锁的线程未必先获取锁，而公平锁可以保证先申请锁的线程一定先获得锁
condition
await()方法会使当前线程等待，同时释放当前锁，当其地接程中使用signal0时成者signalAll0方法时，线程会重新获得锁并把继续执行。或者当线程别中断时，也能跳出等待，这和Object.wait0方法很相似。
singal()方法用于唤醒一个在等待中的线程，相对的singalAll()方法会唤醒所有等待的线程。这和Obejct.notify0方法很类似。
class ReentrantLockTest implements Runnable{
private static ReentrantLock reentrantLock = new ReentrantLock();
private static Condition condition = reentrantLock.newCondition();
@Override
public void run() {
try {
reentrantLock.lock();
condition.await();
System.out.println("thread is going on");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
reentrantLock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
Thread thread1 = new Thread(reentrantLockTest);
thread1.start();
//让主线程睡2秒，然后进行condition.signal();通知子线程继续执行
Thread.sleep(2000);
reentrantLock.lock();
condition.signal();
reentrantLock.unlock();
thread1.join();
System.out.println("end");
}
}
信号量
Semaphore是一个并发工具类，用来控制可同时并发的线程数，其内部维护了一组虚拟许可，通过构造器指定许可的数量，每次线程执行操作时先通过acquire方法获得许可，执行完毕再通过release方法释放许可。如果无可用许可，那么acquire方法将一直阻塞，直到其它线程释放许可。
对比线程池
使用Seamphore，你创建了多少线程，实际就会有多少线程进行执行，只是可同时执行的线程数量会受到限制。但使用线程池，你创建的线程只是作为任务提交给线程池执行，实际工作的线程由线程池创建，并且实际工作的线程数量由线程池自己管理。
参考：https://blog.csdn.net/mryang125/article/details/81490783
class ReentrantLockTest implements Runnable{
private final static Semaphore signal = new Semaphore(5);
@Override
public void run() {
try {
signal.acquire();
//模拟耗时操作
System.out.println("线程 "+Thread.currentThread().getId()+" done");
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
signal.release();
}
}
public static void main(String[] args) throws InterruptedException {
//主线程会中断
//ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
//Thread thread1=null;
//for (int i=0;i<20;i++){
//thread1 = new Thread(reentrantLockTest);
//thread1.start();
//}
//thread1.join();
//System.out.println("end");
//主线程不会中断
ExecutorService executorService = Executors.newFixedThreadPool(20);
ReentrantLockTest reentrantLockTest1 = new ReentrantLockTest();
for (int i=0;i<20;i++){
executorService.submit(reentrantLockTest1);
}
System.out.println("end");
}
}
ReadWriteLock
读读之间不会堵塞
读会堵塞写，写也会堵塞读
写写之间堵塞
ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock(); ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock(); ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
CountDownLatch
class T implements Runnable{
static final CountDownLatch countDownLatch = new CountDownLatch(10);
private final static T t = new T();
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("等待子线程 "+Thread.currentThread().getId()+" 执行完毕");
countDownLatch.countDown();
System.out.println(countDownLatch.getCount());
}
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i=0;i<10;i++){
executorService.submit(t);
}
countDownLatch.await();
System.out.println("所有的子线程执行完毕,主线程继续执行");
executorService.shutdown();
}
}
CyclicBarrier
CountDownLatch 同步计数器，主要用于线程间的控制，但计数无法被重置，如果需要重置计数，请考虑使用 CyclicBarrier 。
可以循环复用，比CountDownLatch功能更加强大
class T{
static CyclicBarrier cyclicBarrier = new CyclicBarrier(10, new BarrierRun(false));
public static class soldier implements Runnable{
String soldierName;
public soldier(String soldierName) {
this.soldierName=soldierName;
}
@Override
public void run() {
try {
//等待所有的士兵(线程)到齐
System.out.println("士兵 "+soldierName+" 到齐");//执行10次
//所有的线程到达完毕之后,执行一次CyclicBarrier(int parties, Runnable barrierAction)中barrierAction的run方法
cyclicBarrier.await();//执行1次
System.out.println("士兵 "+soldierName+" 开始工作");
//等待所有的士兵(线程)到齐
cyclicBarrier.await();
System.out.println("士兵 "+soldierName+" 继续开始工作");
cyclicBarrier.await();
//可以继续await()
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
public static class BarrierRun implements Runnable{
boolean flag;
public BarrierRun( boolean flag) {
this.flag = flag;
}
@Override
public void run() {
if (flag){
System.out.println("所有士兵完成任务");
}else{
System.out.println("所有的士兵集合完毕");
this.flag = true;
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread[] threads = new Thread[10];
for (int i = 0; i < 10; i++) {
String s = String.valueOf(i);
threads[i] = new Thread(new soldier(s));
threads[i].start();
}
}
}
LockSupport
底层实现
线程中断不会抛出异常，能够响应中断，但不抛出异常。
中断响应的结果是，park0函数的返回，可以从Thread.interupted0想到中断标志
测试过程中：如果我中断thread1不知道为什么
class T{
private static Object object = new Object();
public static class ChangeObjectThread extends Thread{
public ChangeObjectThread(String name) {
super.setName(name);
}
@Override
public void run() {
synchronized (object){
System.out.println("in "+getName());
//如果LockSupport没有获取许可(unpark),就将该线程挂起
LockSupport.park();
}
}
}
public static void main(String[] args) throws InterruptedException {
ChangeObjectThread thread1 = new ChangeObjectThread("thread1");
thread1.start();
ChangeObjectThread thread2 = new ChangeObjectThread("thread2");
thread2.start();
//无论线程先执行unpark还是先执行park,线程都不会堵塞(unpark让该线程获取一个许可)
LockSupport.unpark(thread1);
LockSupport.unpark(thread2);
System.out.println("unpark不会堵塞unpark");
}
}
BlockingQueue
接口
性能不高
参考：https://www.cnblogs.com/jackyuj/archive/2010/11/24/1886553.html
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* @author jackyuj
*/
public class BlockingQueueTest {
public static void main(String[] args) throws InterruptedException {
// 声明一个容量为10的缓存队列
BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
Producer producer1 = new Producer(queue);
Producer producer2 = new Producer(queue);
Producer producer3 = new Producer(queue);
Consumer consumer = new Consumer(queue);
// 借助Executors
ExecutorService service = Executors.newCachedThreadPool();
// 启动线程
service.execute(producer1);
service.execute(producer2);
service.execute(producer3);
service.execute(consumer);
// 执行10s
Thread.sleep(10 * 1000);
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(2000);
// 退出Executor
service.shutdown();
}
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* 消费者线程
*
* @author jackyuj
*/
public class Consumer implements Runnable {
public Consumer(BlockingQueue<String> queue) {
this.queue = queue;
}
public void run() {
System.out.println("启动消费者线程！");
Random r = new Random();
boolean isRunning = true;
try {
while (isRunning) {
System.out.println("正从队列获取数据...");
String data = queue.poll(2, TimeUnit.SECONDS);
if (null != data) {
System.out.println("拿到数据：" + data);
System.out.println("正在消费数据：" + data);
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
} else {
// 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。
isRunning = false;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出消费者线程！");
}
}
private BlockingQueue<String> queue;
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生产者线程
*
* @author jackyuj
*/
public class Producer implements Runnable {
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
String data = null;
Random r = new Random();
System.out.println("启动生产者线程！");
try {
while (isRunning) {
System.out.println("正在生产数据...");
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
data = "data:" + count.incrementAndGet();
System.out.println("将数据：" + data + "放入队列...");
if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
System.out.println("放入数据失败：" + data);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出生产者线程！");
}
}
public void stop() {
isRunning = false;
}
private volatile boolean isRunning = true;
private BlockingQueue queue;
private static AtomicInteger count = new AtomicInteger();
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
ConcurrentLinkedQueue
高性能线程间通讯
ConcurrentLinkedQueue concurrentLinkedQueue = new ConcurrentLinkedQueue();
concurrentLinkedQueue.add("s");
Object remove = concurrentLinkedQueue.remove();
对于高并发我们还可以采用CAS无锁
乐观锁的实现原理，当前版本号是不是我预期的版本号，如果是就进行修改，并且修改是原子性。
CAS算法即是：Compare And Swap,比较并且替换；
CAS算法存在着三个参数，内存值V，旧的预期值A，以及要更新的值B。当且仅当内存值V和预期值B相等的时候，才会将内存值修改为B，否则什么也不做，直接返回false;
比如说某一个线程要修改某个字段的值，当这个值初始化的时候会在内存中完成，根据Java内存模型，该线程保存着这个变量的一个副本；当且仅当这个变量的副本和内存的值如果相同，那么就可以完成对值得修改，并且这个CAS操作完全是原子性的操作，也就是说此时这个操作不可能被中断。
CAS只能对一个变量进行原子操作，如果需要的变量需要同时进行的话，有一个办法就是将变量打包。
CAS存在ABA的问题：假设线程1从内存中取出了A，线程2也从内存中取出了A，并且将值修改为B，最后又改为A，当线程1去更新值得时候发现内存中的数据和线程备份数据相同，可以更新；但是此时内存中的值其实发生了变化的，只不过又变回去了；
参考：https://www.cnblogs.com/gosaint/p/9045494.html
AtomicInteger atomicInteger = new AtomicInteger(1); int i = atomicInteger.addAndGet(1);
内部实现：Unsafe.getUnsafe().getAndAddInt(this, valueOffset, delta) + delta;
如何实现高并发无锁缓存？
参考：https://blog.csdn.net/xybelieve1990/article/details/70313076
java代码实现
首先对1条数据添加一个字段（version，lastTime等），用于进行乐观锁控制。对1条数据进行修改，首先获取这个数据（获取version，lastTime），如果想对这条数据进行修改的时候，version或者lastTime需要作为where查询条件。
