玩转JUC工具,Java并发编程不再危机四伏

简介: 当今互联网应用普遍需要支持高并发访问,而Java作为一种广泛使用的编程语言,其并发编程能力对于实现高性能的应用非常重要。而Java的JUC(java.util.concurrent)并发工具就提供了许多实用的工具类和接口,可以让Java应用轻松实现高效的并发编程。

前言
  当今互联网应用普遍需要支持高并发访问,而Java作为一种广泛使用的编程语言,其并发编程能力对于实现高性能的应用非常重要。而Java的JUC(java.util.concurrent)并发工具就提供了许多实用的工具类和接口,可以让Java应用轻松实现高效的并发编程。
ReetrantLock
  ReentrantLock是Java提供的一个可重入锁,也是Java并发编程中最常用的一种锁。与synchronized关键字相比,ReentrantLock具有更大的灵活性和功能,可以更好地支持并发编程的实现。
特点

可重入性:与synchronized一样,ReentrantLock也支持可重入锁,即同一个线程可以重复获得该锁,而不会出现死锁。
公平锁:ReentrantLock可以创建公平锁,即按照线程请求的顺序分配锁,确保等待时间最长的线程最先获得锁,避免了线程饥饿问题。
中断响应:当线程等待获取锁的过程中,可以通过调用interrupt()方法中断等待,避免线程无限等待。
条件变量:ReentrantLock可以创建多个Condition对象,用于控制线程等待和唤醒,从而实现更灵活的线程协作。
性能优越:在高度竞争的多线程环境中,ReentrantLock相比synchronized有更好的性能表现,特别是在多处理器系统中。

简单使用
模拟秒杀商品场景
public class SecKillDemo {

private static int stock = 1;

// 秒杀锁
private static final ReentrantLock lock = new ReentrantLock();

public static void main(String[] args) {
    // 模拟多个用户抢购
    for (int i = 0; i < 10; i++) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                // 加锁
                lock.lock();
                try {
                    // 判断库存是否足够
                    if (stock > 0) {
                        // 模拟生成订单
                        System.out.println(Thread.currentThread().getName() + "抢购成功,生成订单");
                        // 减少库存
                        stock--;
                    } else {
                        System.out.println(Thread.currentThread().getName() + "抢购失败,库存不足");
                    }
                } finally {
                    // 解锁
                    lock.unlock();
                }
            }
        }).start();
    }
}

}
复制代码
加锁结果:最先进来的线程抢购成功其余线程抢购失败

不加锁结果:4个线程抢购成功出现超卖现象

可重入示例
  当一个线程获得了ReentrantLock锁之后,如果该线程继续请求获得这个锁,那么该线程可以继续获得这个锁,这种情况就是ReentrantLock锁的可重入性。下面是一个表现出ReentrantLock锁的可重入性的例子:
public class ReentrantLockExample {

private final ReentrantLock lock = new ReentrantLock();

public void outer() {
    lock.lock(); // 第一次获取锁
    try {
        inner();
    } finally {
        lock.unlock(); // 释放锁
    }
}

public void inner() {
    lock.lock(); // 第二次获取锁
    try {
        System.out.println("第二次获取锁");
    } finally {
        lock.unlock(); // 释放锁
    }
}

public static void main(String[] args) {
    ReentrantLockExample example = new ReentrantLockExample();
    example.outer();
}

}
复制代码
  在这个例子中,我们定义了一个ReentrantLock锁,并创建了一个outer方法和一个inner方法。在outer方法中,我们首先通过调用lock方法获取锁,并在try块中调用inner方法,然后在finally块中释放锁。在inner方法中,我们再次获取锁,并输出一条信息。
  可以看到,在outer方法中,我们第一次获取锁,然后调用inner方法,inner方法中又通过lock方法获取了锁,但这次获取锁是成功的,并且能够正常输出信息,说明ReentrantLock锁具有可重入性。
Condition
  当使用ReentrantLock时,可以使用Condition对象来进行线程的协调和等待。
public class ConditionDemo {

private ReentrantLock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
private int count = 0;

public void increment() {
    lock.lock();
    try {
        while (count == 1) {  // 如果计数器已经达到1,就等待
            condition.await();
        }
        count++;  // 计数器加1
        System.out.println(Thread.currentThread().getName() + ": " + count);
        condition.signalAll();  // 唤醒其他等待的线程
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
}

public void decrement() {
    lock.lock();
    try {
        while (count == 0) {  // 如果计数器已经为0,就等待
            condition.await();
        }
        count--;  // 计数器减1
        System.out.println(Thread.currentThread().getName() + ": " + count);
        condition.signalAll();  // 唤醒其他等待的线程
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
}

public static void main(String[] args) {
    ConditionDemo demo = new ConditionDemo();

    // 创建两个线程分别进行增加和减少操作
    new Thread(() -> {
        for (int i = 0; i < 10; i++) {
            demo.increment();
        }
    }, "Thread-A").start();

    new Thread(() -> {
        for (int i = 0; i < 10; i++) {
            demo.decrement();
        }
    }, "Thread-B").start();
}

}
复制代码
  在这个示例中,有一个计数器count,两个线程分别进行增加和减少操作。在increment()和decrement()方法中,通过调用ReentrantLock的lock()方法获取锁,并在执行操作之前使用while循环判断条件是否满足,如果不满足就通过调用Condition的await()方法等待。
  当条件满足时,线程执行相应的操作,并使用Condition的signalAll()方法唤醒其他等待的线程。在这个示例中,当计数器为1时,增加线程就会等待,直到计数器减为0;当计数器为0时,减少线程就会等待,直到计数器增加为1。
应用场景

解决多线程竞争资源的问题,例如多个线程同时对同一个数据库进行写操作,可以使用ReentrantLock保证每次只有一个线程能够写入。

实现多线程任务的顺序执行,例如在一个线程执行完某个任务后,再让另一个线程执行任务。

实现多线程等待/通知机制,例如在某个线程执行完某个任务后,通知其他线程继续执行任务

Semaphore
  Semaphore(信号量)是JUC并发工具包中的一种同步工具,用于管理一个或多个共享资源的访问。Semaphore 维护一个计数器,该计数器可以对共享资源的访问进行控制,类似于停车场的车位管理,当所有的车位已满时,新来的车辆必须等待其他车辆离开才能进入停车场。
Semaphore实现限流
public class RateLimiter {

private Semaphore semaphore;

public RateLimiter(int permits) {
    semaphore = new Semaphore(permits);
}

public boolean tryAcquire() {
    return semaphore.tryAcquire();
}

public void release() {
    semaphore.release();
}

}
复制代码
  在上述代码中,我们定义了一个名为RateLimiter的限流器类。它有一个构造函数,用于初始化Semaphore计数器,其中参数permits表示允许同时访问的线程数。
  在tryAcquire()方法中,我们使用了Semaphore的tryAcquire()方法来尝试获取一个许可,如果获取成功则返回true,否则返回false。
  在release()方法中,我们调用Semaphore的release()方法来释放一个许可。
public static void main(String[] args) {

int permits = 5;
RateLimiter rateLimiter = new RateLimiter(permits);

// 模拟10个线程并发请求
for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        try {
            if (rateLimiter.tryAcquire()) {
                System.out.println(Thread.currentThread().getName() + " 获得访问权限");
                Thread.sleep(1000);
            } else {
                System.out.println(Thread.currentThread().getName() + " 被限流");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            rateLimiter.release();
            System.out.println(Thread.currentThread().getName() + " 释放访问权限");
        }
    }, "Thread-" + (i + 1)).start();
}

}
复制代码
在上述代码中,我们创建了一个限流器RateLimiter,允许同时有5个线程访问。然后,我们模拟10个请求。当有可用许可时,请求被允许访问;当所有许可都被占用时,请求被拒绝访问。
应用场景

限流:Semaphore可以用于限制对共享资源的并发访问数量,以控制系统的流量。

资源池:Semaphore可以用于实现资源池,以维护一组有限的共享资源。

CountDownLatch
  CountDownLatch可以帮助控制线程之间的执行顺序。在某些场景下,可能需要等待多个线程执行完毕后,再继续执行某些操作,这时候就可以使用CountDownLatch来实现线程的等待。
  CountDownLatch内部维护了一个计数器,该计数器的初始值可以通过构造函数进行指定。在主线程中调用CountDownLatch的await()方法会使当前线程等待,直到计数器的值为0时才会继续执行。而在其他线程中调用CountDownLatch的countDown()方法则会将计数器的值减1。当计数器的值减到0时,之前在主线程中调用await()方法的线程就会继续执行。
CountDownLatch实现多任务合并
  比如说,有一个系统需要进行批量数据导入,数据是从多个文件中读取的,每个文件需要一个线程进行处理,等所有文件处理完毕后再进行下一步操作,这时候就可以使用CountDownLatch来实现等待多个线程执行完毕后再继续执行。。
public class BatchImportDemo {

private CountDownLatch latch;
private String[] filenames = {"file1.txt", "file2.txt", "file3.txt"};

public BatchImportDemo() {
    this.latch = new CountDownLatch(filenames.length);
}

public void start() {
    for (String filename : filenames) {
        new Thread(new ImportTask(filename, latch)).start();
    }
    try {
        latch.await(); // 等待所有线程执行完毕
        System.out.println("所有文件处理完毕,开始进行下一步操作。");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

private static class ImportTask implements Runnable {
    private String filename;
    private CountDownLatch latch;

    public ImportTask(String filename, CountDownLatch latch) {
        this.filename = filename;
        this.latch = latch;
    }

    @Override
    public void run() {
        // 处理文件的逻辑
        System.out.println("正在处理文件 " + filename);
        try {
            Thread.sleep(2000); // 模拟文件处理
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("文件 " + filename + " 处理完毕。");
        latch.countDown(); // 计数器减1
    }
}

public static void main(String[] args) {
    BatchImportDemo demo = new BatchImportDemo();
    demo.start();
}

}
复制代码
  在这个示例中,我们定义了一个BatchImportDemo类,它包含了一个CountDownLatch实例和一个字符串数组filenames,表示需要处理的文件名。在start()方法中,我们启动了多个线程来处理每个文件,并调用了latch.await()方法来等待所有线程执行完毕。而在每个线程中,我们执行了具体的文件处理逻辑,并在处理完毕后调用了latch.countDown()方法来将计数器减1。
  运行上述代码,输出结果为
正在处理文件 file1.txt
正在处理文件 file2.txt
正在处理文件 file3.txt
文件 file1.txt 处理完毕。
文件 file2.txt 处理完毕。
文件 file3.txt 处理完毕。
所有文件处理完毕,开始进行下一步操作。
复制代码
  可以看到,当所有文件处理完毕后,程序输出了“所有文件处理完毕,开始进行下一步操作。”的信息。这说明我们成功地使用CountDownLatch来等待多个线程执行完毕后再进行后续的操作。
应用场景

主线程等待多个线程执行完毕后再继续执行。
多个线程等待某个操作完成后再继续执行。
实现并发任务的协调,例如多个线程同时执行不同的子任务,当所有子任务都执行完毕后,再执行后续的操作。

CyclicBarrier
  CyclicBarrier(回环栅栏或循环屏障),是 Java 并发库中的一个同步工具,通过它可以实现让一组线程等待至某个状态(屏障点)
之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后,CyclicBarrier可以被重用。
CyclicBarrier多个线程协同工作
public class CyclicBarrierDemo {

public static void main(String[] args) throws InterruptedException {
    int threadCount = 5;
    CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount, new Runnable() {
        @Override
        public void run() {
            System.out.println("所有线程执行完成,开始执行主线程...");
        }
    });
    for (int i = 0; i < threadCount; i++) {
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 执行任务...");
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + " 执行任务完成...");
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }).start();
    }
    System.out.println("主线程执行...");
}

}
复制代码
  上述示例中,定义了5个线程,这5个线程需要等待所有线程都执行完成后,才能继续执行主线程。在定义CyclicBarrier时,将屏障点的数量设置为5,当所有线程都到达屏障点时,会执行Runnable中的任务,输出 "所有线程执行完成,开始执行主线程..."。每个线程执行任务完成后,会调用CyclicBarrier的await()方法,等待其他线程执行完成。
  执行以上代码,输出如下:
Thread-0 执行任务...
Thread-1 执行任务...
Thread-2 执行任务...
Thread-3 执行任务...
Thread-4 执行任务...
主线程执行...
Thread-1 执行任务完成...
Thread-0 执行任务完成...
Thread-3 执行任务完成...
Thread-2 执行任务完成...
Thread-4 执行任务完成...
所有线程执行完成,开始执行主线程...
复制代码
  从输出结果可以看出,所有线程都先执行各自的任务,然后等待其他线程执行完成,当所有线程都执行完成后,执行Runnable中的任务,输出 "所有线程执行完成,开始执行主线程...",最后主线程继续执行。
应用场景

多线程任务:CyclicBarrier 可以用于将复杂的任务分配给多个线程执行,并在所有线程完成工作后触发后续作。

数据处理:CyclicBarrier 可以用于协调多个线程间的数据处理,在所有线程处理完数据后触发后续操作。

Phaser
  Phaser用于协调多个线程的执行。它提供了一些方便的方法来管理多个阶段的执行,可以让程序员灵活地控制线程的执行顺序和阶段性的执行。Phaser可以被视为CyclicBarrier和CountDownLatch的进化版,它能够自适应地调整并发线程数,可以动态地增加或减少参与线程的数量。所以Phaser特别适合使用在重复执行或者重用的情况。
Phaser阶段任务使用
public class PhaserDemo {

public static void main(String[] args) {
    int numPhases = 3; // 设置阶段数为3
    int numThreads = 5; // 设置线程数为5
    Phaser phaser = new Phaser(numThreads); // 创建Phaser对象并注册线程数
    for (int i = 0; i < numThreads; i++) {
        new Thread(new Worker(phaser)).start(); // 创建并启动线程
    }
    for (int i = 0; i < numPhases; i++) {
        System.out.println("Phase " + i + " 阶段开始");
        phaser.arriveAndAwaitAdvance(); // 等待所有线程到达同步点
        System.out.println("Phase " + i + " 阶段完成");
    }
}

static class Worker implements Runnable {
    private Phaser phaser;

    Worker(Phaser phaser) {
        this.phaser = phaser;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " 开始执行");
        for (int i = 0; i < 3; i++) { // 模拟每个线程需要执行3个任务
            phaser.arriveAndAwaitAdvance(); // 到达同步点并等待其他线程
            System.out.println(Thread.currentThread().getName() + " 完成第" + i + "个任务");
        }
    }
}

}
复制代码
  上述示例中,我们创建了一个Phaser对象,将线程数和阶段数分别设置为5和3,然后创建5个线程并启动它们。每个线程需要完成3个任务,在完成每个任务后调用arriveAndAwaitAdvance()方法到达同步点并等待其他线程,等到所有线程到达同步点后才会进入下一个阶段。最终,程序输出了每个线程完成任务的信息,以及每个阶段的开始和结束时间。
应用场景

多线程执行多阶段任务,需要协调各个线程的执行顺序。
多线程进行游戏或模拟操作,需要协调各个线程的执行时机。
多个线程需要协同工作来处理一个大型问题,例如搜索算法或者数据分析等。

Exchanger
  Exchanger是JUC(java.util.concurrent)并发工具之一,它提供了一个同步点,使得两个线程可以交换对象。Exchanger中交换对象的过程是一个阻塞方法,只有在两个线程都到达同步点时,才会交换对象,并且在交换完成后,两个线程会继续执行自己的代码。
  Exchanger通常用于实现数据的同步和线程间的通信,例如在生产者和消费者模式中,可以使用Exchanger来实现生产者和消费者之间的数据交换。另外,Exchanger也可以用于遗传算法、数据加密等应用场景中。
Exchanger使用
public class ExchangerDemo {

public static void main(String[] args) {
    Exchanger<String> exchanger = new Exchanger<>();
    Thread thread1 = new Thread(() -> {
        try {
            String data1 = "data1";
            String data2 = exchanger.exchange(data1);
            System.out.println("Thread1 received: " + data2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    Thread thread2 = new Thread(() -> {
        try {
            String data2 = "data2";
            String data1 = exchanger.exchange(data2);
            System.out.println("Thread2 received: " + data1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    thread1.start();
    thread2.start();

}

}
复制代码
  在上述示例代码中,创建了一个Exchanger对象exchanger,然后启动了两个线程thread1和thread2。线程thread1将字符串"data1"交换给线程thread2,并且接收到线程thread2交换过来的字符串"data2";线程thread2将字符串"data2"交换给线程thread1,并且接收到线程thread1交换过来的字符串"data1"。最后,两个线程都输出了接收到的数据。
应用场景

生产者消费者模式:在生产者和消费者模式中,可以使用Exchanger来实现生产者和消费者之间的数据交换,从而实现数据的同步和交流。
遗传算法:在遗传算法中,可以使用Exchanger来实现父代和子代之间的基因交换,从而实现基因的进化和优化。
数据加密:在数据加密中,可以使用Exchanger来实现加密和解密数据之间的交换,从而保证数据的安全性。
线程间协作:在需要多个线程协作完成某项任务的场景中,可以使用Exchanger来实现线程间的数据交换和同步,从而协同完成任务。

  需要注意的是,Exchanger只适用于两个线程之间的数据交换,如果需要多个线程之间的数据交换和同步,可以使用其他的并发工具,例如CyclicBarrier、CountDownLatch等。在选择并发工具时,应根据具体的场景和需求来进行选择。

相关文章
|
2月前
|
安全 Java
多线程(进阶三:JUC)
多线程(进阶三:JUC)
45 0
|
6月前
|
缓存 安全 算法
JUC第一讲:juc并发包深入理解(P6熟练 P7精通)
JUC第一讲:juc并发包深入理解(P6熟练 P7精通)
|
9月前
|
Web App开发 安全 Java
JUC高并发编程(一)——JUC基础知识
JUC高并发编程(一)——JUC基础知识
101 0
|
10月前
|
安全 Java Linux
JUC (java并发编程学习分享篇)
JUC (java并发编程学习分享篇)
44 0
|
11月前
【JUC基础】02. JUC思维导图
juc思维导图,用于JUC内容学习
|
11月前
|
消息中间件 资源调度 Java
【JUC基础】01. 初步认识JUC
前段时间,有朋友跟我说,能否写一些关于JUC的教程文章。本来呢,JUC也有在我的专栏计划之内,只是一直都还没空轮到他,那么既然有这样的一个契机,那就把JUC计划提前吧。那么今天就重点来初步认识一下什么是JUC,以及一些基本的JUC相关基础知识。
126 0
【JUC基础】01. 初步认识JUC
|
设计模式 Java 数据处理
【Java并发编程系列8】多线程实战
Java多线程的学习,也有大半个月了,从开始学习Java多线程时,就给自己定了一个小目标,希望能写一个多线程的Demo,今天主要是兑现这个小目标。
600 0
|
存储 SQL 缓存
JUC 并发编程学习笔记(总)
JUC 并发编程学习笔记(总)
73 0
JUC 并发编程学习笔记(总)
|
Java 调度 C++
JUC 并发编程学习笔记(上)
JUC 并发编程学习笔记(上)
65 0
JUC 并发编程学习笔记(上)
|
缓存 Java API
JUC 并发编程学习笔记(中)
JUC 并发编程学习笔记(中)
60 0
JUC 并发编程学习笔记(中)