Java中的多线程与并发编程
今天我们来探讨一下Java中的多线程与并发编程。多线程编程可以充分利用CPU资源，提高程序的运行效率。然而，由于多线程编程的复杂性，开发过程中需要特别注意线程安全和死锁等问题。本文将介绍Java多线程的基本概念、线程的创建与管理、同步机制以及常见的并发工具类。
一、多线程的基本概念
在单线程环境中，一个时间点只能有一个任务执行，多线程则允许多个任务同时执行，从而提高系统的吞吐量。Java中的多线程是通过java.lang.Thread类和java.lang.Runnable接口来实现的。
二、线程的创建与管理
- 继承Thread类
package cn.juwatech.multithreading;
public class MyThread extends Thread {
@Override
public void run() {
System.out.println("Thread is running");
}
public static void main(String[] args) {
MyThread thread = new MyThread();
thread.start();
}
}- 实现Runnable接口
package cn.juwatech.multithreading;
public class MyRunnable implements Runnable {
@Override
public void run() {
System.out.println("Runnable is running");
}
public static void main(String[] args) {
Thread thread = new Thread(new MyRunnable());
thread.start();
}
}- 使用Callable和Future
Callable接口类似于Runnable，但它可以返回结果并且可以抛出异常。Future接口表示异步计算的结果。
package cn.juwatech.multithreading;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class MyCallable implements Callable<String> {
@Override
public String call() throws Exception {
return "Callable result";
}
public static void main(String[] args) {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(new MyCallable());
try {
String result = future.get();
System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
executor.shutdown();
}
}
}三、线程同步
多线程环境下，多个线程同时访问共享资源可能导致数据不一致的问题。Java提供了多种同步机制来解决这个问题。
- 同步方法和同步块
package cn.juwatech.multithreading;
public class SynchronizedExample {
private int count = 0;
public synchronized void increment() {
count++;
}
public void incrementWithSyncBlock() {
synchronized (this) {
count++;
}
}
public static void main(String[] args) {
SynchronizedExample example = new SynchronizedExample();
for (int i = 0; i < 1000; i++) {
new Thread(example::increment).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + example.count);
}
}- 使用ReentrantLock
ReentrantLock提供了比sychronized更灵活的锁机制。
package cn.juwatech.multithreading;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class LockExample {
private final Lock lock = new ReentrantLock();
private int count = 0;
public void increment() {
lock.lock();
try {
count++;
} finally {
lock.unlock();
}
}
public static void main(String[] args) {
LockExample example = new LockExample();
for (int i = 0; i < 1000; i++) {
new Thread(example::increment).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + example.count);
}
}四、线程间通信
- 使用wait()、notify()和notifyAll()
package cn.juwatech.multithreading;
public class WaitNotifyExample {
private final Object lock = new Object();
public void produce() throws InterruptedException {
synchronized (lock) {
System.out.println("Producer is producing...");
lock.wait();
System.out.println("Producer resumed.");
}
}
public void consume() throws InterruptedException {
Thread.sleep(1000);
synchronized (lock) {
System.out.println("Consumer is consuming...");
lock.notify();
}
}
public static void main(String[] args) {
WaitNotifyExample example = new WaitNotifyExample();
Thread producer = new Thread(() -> {
try {
example.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
Thread consumer = new Thread(() -> {
try {
example.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
producer.start();
consumer.start();
}
}- 使用BlockingQueue
BlockingQueue提供了线程安全的生产者-消费者模式。
package cn.juwatech.multithreading;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class BlockingQueueExample {
private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
public static void main(String[] args) {
Thread producer = new Thread(() -> {
try {
for (int i = 0; i < 100; i++) {
queue.put(i);
System.out.println("Produced: " + i);
Thread.sleep(50);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
Thread consumer = new Thread(() -> {
try {
while (true) {
int value = queue.take();
System.out.println("Consumed: " + value);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
producer.start();
consumer.start();
}
}五、并发工具类
Java的java.util.concurrent包提供了丰富的并发工具类，简化了多线程编程。
- CountDownLatch
CountDownLatch允许一个或多个线程等待其他线程完成操作。
package cn.juwatech.multithreading;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
private static final int TASK_COUNT = 5;
public static void main(String[] args) {
CountDownLatch latch = new CountDownLatch(TASK_COUNT);
for (int i = 0; i < TASK_COUNT; i++) {
new Thread(() -> {
try {
System.out.println("Task started.");
Thread.sleep((long) (Math.random() * 1000));
System.out.println("Task completed.");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}).start();
}
try {
latch.await();
System.out.println("All tasks completed.");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}- CyclicBarrier
CyclicBarrier允许一组线程互相等待，直到到达公共屏障点。
package cn.juwatech.multithreading;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
private static final int THREAD_COUNT = 3;
public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT, () -> {
System.out.println("All threads have reached the barrier.");
});
for (int i = 0; i < THREAD_COUNT; i++) {
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " is waiting at the barrier.");
barrier.await();
System.out.println(Thread.currentThread().getName() + " has crossed the barrier.");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}六、线程池
线程池通过重用线程来提高性能，避免频繁创建和销毁线程的开销。Java提供了ExecutorService来管理线程池。
package cn.juwatech.multithreading;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10
; i++) {
executor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is executing a task.");
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executor.shutdown();
}
}
