JUC是JAVA Util Concurrency的缩写,即Java并发工具包。JUC包提供了一些常用的线程和并发编程工具类,帮助开发者更方便地开发多线程应用程序,提高程序的并发性能。JUC包的主要特点包括:
在多线程环境下,由于多个线程同时访问同一个变量可能会导致数据不一致的问题。原子操作类可以保证在多线程环境下对变量的操作是原子性的,即不会出现线程安全问题。
JJUC包中提供了以下几个原子操作类:
这些原子操作类都提供了一系列的方法,如get、set、addAndGet、compareAndSet等,可以实现对变量的原子操作。值得注意的是,使用原子操作类并不能解决所有的线程安全问题,需要根据具体情况进行判断和选择。
AtomicInteger用于对int类型的变量进行原子操作。
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerDemo {
private static AtomicInteger count = new AtomicInteger(0);
public static void mAIn(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
count.getAndIncrement();
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + count.get());
}
}
AtomicLong用于对long类型的变量进行原子操作。
import java.util.concurrent.atomic.AtomicLong;
public class AtomicLongDemo {
private static AtomicLong count = new AtomicLong(0);
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
count.getAndIncrement();
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + count.get());
}
}
AtomicBoolean用于对boolean类型的变量进行原子操作。
import java.util.concurrent.atomic.AtomicBoolean;
public class AtomicBooleanDemo {
private static AtomicBoolean flag = new AtomicBoolean(true);
public static void main(String[] args) {
new Thread(() -> {
while (flag.get()) {
System.out.println("Running...");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
flag.set(false);
System.out.println("Stopped.");
}
}
AtomicIntegerArray用于对int数组中的元素进行原子操作。
import java.util.concurrent.atomic.AtomicIntegerArray;
public class AtomicIntegerArrayDemo {
private static AtomicIntegerArray arr = new AtomicIntegerArray(new int[]{0, 0});
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
arr.getAndIncrement(j % 2);
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedExceptione) {
e.printStackTrace();
}
System.out.println("Array: " + arr);
}
}
AtomicReference用于对引用类型的变量进行原子操作。
import java.util.concurrent.atomic.AtomicReference;
public class AtomicReferenceDemo {
static class Person {
String name;
int age;
public Person(String name, int age) {
this.name = name;
this.age = age;
}
@Override
public String toString() {
return "Person{name='" + name + "', age=" + age + "}";
}
}
private static AtomicReference<Person> personRef = new AtomicReference<>(new Person("Tom", 18));
public static void main(String[] args) {
Person oldPerson = personRef.get();
Person newPerson = new Person("Jerry", 20);
if (personRef.compareAndSet(oldPerson, newPerson)) {
System.out.println("Update success, old value: " + oldPerson + ", new value: " + newPerson);
} else {
System.out.println("Update failed.");
}
System.out.println("Person: " + personRef.get());
}
}
同步队列类是一种特殊的队列,它可以在多线程环境下实现数据的生产和消费过程的同步。JUC包中提供了以下几个同步队列类:
这些同步队列类提供了一系列的方法,如put、take、offer、poll等,可以实现对队列的操作。同步队列类还提供了一些扩展方法,如drainTo、peek等。
同步队列类的特点在于它们可以实现生产者-消费者模式。多个线程可以同时往队列中添加元素或者同时从队列中取出元素,当队列为空或者已满时,线程会被阻塞,直到有其他线程进行相应的操作。这种机制可以有效地控制线程间的同步和协作,避免了线程间的竞争和死锁问题。
使用同步队列类时需要注意以下几点:
ArrayBlockingQueue是一个有界队列,它的容量是固定的。当队列已满时,添加元素的线程会被阻塞,直到有其他线程取出元素后才能继续添加。
import java.util.concurrent.ArrayBlockingQueue;
public class ArrayBlockingQueueDemo {
private static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 100; i++) {
try {
queue.put(i);
System.out.println("Producer: " + i);
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
LinkedBlockingQueue是一个无界队列,它的容量是不限制的。当队列为空时,取出元素的线程会被阻塞,直到有其他线程添加元素后才能继续取出。
import java.util.concurrent.LinkedBlockingQueue;
public class LinkedBlockingQueueDemo {
private static LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 100; i++) {
try {
queue.put(i);
System.out.println("Producer: " + i);
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
SynchronousQueue是一个没有缓冲的队列,它的每个插入操作必须等待另一个线程执行相应的删除操作,反之亦然。当队列中有一个元素时,插入操作会被阻塞,直到有其他线程取出元素后才能继续插入。
import java.util.concurrent.SynchronousQueue;
public class SynchronousQueueDemo {
private static SynchronousQueue queue = new SynchronousQueue<>();
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
System.out.println("Producer: " + i);
queue.put(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
CountDownLatch是一种同步工具类,它可以使一个或多个线程等待另一组线程完成操作后再继续执行。CountDownLatch的作用类似于“计数器”,在初始化时设置一个计数值,每当一个线程完成任务后就将计数值减1,当计数值变为0时,等待线程就会被唤醒。
CountDownLatch类提供了两个主要方法:
使用CountDownLatch可以很方便地实现线程间的协作和同步,尤其适用于某些场景下需要等待多个线程都完成某项任务后才能进行下一步操作的情况。
import java.util.concurrent.CountDownLatch;
public class CountDownLatchDemo {
private static CountDownLatch latch = new CountDownLatch(3);
public static void main(String[] args) {
new Thread(() -> {
try {
Thread.sleep(1000);
System.out.println("Thread A finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000);
System.out.println("Thread B finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(3000);
System.out.println("Thread C finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
try {
latch.await();
System.out.println("All threads finished.");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
CyclicBarrier也是一种同步工具类,它可以让一组线程在到达某个屏障点之前相互等待,然后同时执行某个操作。CyclicBarrier的作用类似于“栅栏”,在初始化时设置一个屏障点,每当一个线程到达屏障点时就会被阻塞,直到所有线程都到达屏障点后才会继续执行。
CyclicBarrier类提供了两个主要方法:
使用CyclicBarrier可以很方便地实现一组线程的同步和协作,尤其适用于某些场景下需要多个线程同时开始执行某项任务的情况。
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
private static CyclicBarrier barrier = new CyclicBarrier(3, () -> {
System.out.println("All threads arrived at the barrier.");
});
public static void main(String[] args) {
new Thread(() -> {
try {
Thread.sleep(1000);
System.out.println("Thread A arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000);
System.out.println("Thread B arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(3000);
System.out.println("Thread C arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
信号量是一种经典的并发编程工具,它可以用来限制同时访问某个资源的线程数量。JUC包中提供了以下几个信号量类:
这些信号量类提供了一系列的方法,如acquire、release、tryAcquire等,可以实现对信号量的操作。使用信号量类可以有效地控制线程的并发访问,从而避免竞争和死锁问题。
Semaphore是一个同步工具类,用于控制对公共资源的访问。它通过计数器来实现对资源的访问控制,可以控制同时访问某个资源的线程数量。
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
private static Semaphore semaphore = new Semaphore(2);
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
new Thread(() -> {
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName() + " acquired the semaphore.");
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + " released the semaphore.");
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}
Exchanger是一种同步工具类,它可以使两个线程之间交换数据。Exchanger的作用类似于“交换机”,两个线程分别调用Exchanger对象的exchange方法,将各自持有的数据传递给对方,然后继续执行。
Exchanger类提供了一个exchange方法,可以实现两个线程之间的数据交换。使用Exchanger可以很方便地实现数据在不同线程之间的传递和同步,尤其适用于某些场景下需要进行线程间数据交互的情况。
import java.util.concurrent.Exchanger;
public class ExchangerDemo {
private static Exchanger<String> exchanger = new Exchanger<>();
public static void main(String[] args) {
new Thread(() -> {
try {
String data = "Hello World";
System.out.println("Thread A: before exchange, data = " + data);
data = exchanger.exchange(data);
System.out.println("Thread A: after exchange, data = " + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
String data = "123456789";
System.out.println("Thread B: before exchange, data = " + data);
data = exchanger.exchange(data);
System.out.println("Thread B: after exchange, data = " + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
CompletableFuture是Java8中新增的一个并发工具类,它可以以异步的方式执行任务,并支持任务之间的组合和串联操作。CompletableFuture类的主要特点包括:
CompletableFuture类提供了一系列的方法,如supplyAsync、thenApply、thenAccept、thenCompose等,可以实现对任务的异步执行、组合和串联操作。使用CompletableFuture可以很方便地实现高效、简洁的异步编程方式。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("Task 1 is running.");
return "Result 1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("Task 2 is running.");
return "Result 2";
});
CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, (result1, result2) -> {
System.out.println("Task 3 is running.");
System.out.println("result1: " + result1);
System.out.println("result2: " + result2);
return result1.length() + result2.length();
});
System.out.println("Combined result: " + combinedFuture.get());
}
}
ForkJoin框架是JDK7中引入的一个并行计算框架,它可以将一个大型任务划分为若干个小任务并行执行,然后将各个小任务的结果汇总得到最终结果。ForkJoin框架的主要特点包括:
ForkJoin框架通过ForkJoinPool类来管理线程池和任务调度。使用ForkJoin框架可以很方便地实现高效、简洁的并行计算代码。
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class ForkJoinDemo {
static class Fibonacci extends RecursiveTask<Integer> {
final int n;
Fibonacci(int n) {
this.n = n;
}
protected Integer compute() {
if (n <= 1)
return n;
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
return f2.compute() + f1.join();
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
Fibonacci task = new Fibonacci(10);
int result = pool.invoke(task);
System.out.println(result);
}
}
Java并发编程是一门非常重要的技术,在面对大规模并发处理、高性能计算、分布式系统和云计算等领域时,它扮演着至关重要的角色。本文介绍了Java并发编程中常用的几种并发工具类和框架,包括线程池、锁、原子类、同步队列、同步工具类、CompletableFuture和Fork/Join框架等,并提供了简单的示例代码,希望可以为读者在实践中应用并发编程提供一些参考和启示。