并发编程十大核心模式代码模板

并发编程的核心模式包括10种关键方法，每种都有特定用途和适用场景。1.互斥锁（mutex）通过reentrantlock确保线程对共享资源的独占访问；2.读写锁（readwritelock）允许多个线程同时读取、单个写入，提高读多写少场景效率；3.信号量（semaphore）控制并发访问数量，限制资源使用上限；4.countdownlatch用于等待多个线程任务完成后再继续执行；5.cyclicbarrier使一组线程互相等待到达共同屏障点后继续运行；6.future/callable实现异步任务执行并获取结果；7.线程池（threadpool）复用线程降低创建销毁开销；8.blockingqueue提供线程安全的数据传递机制；9.原子变量（atomic variables）如atomicinteger实现无锁线程安全操作；10.fork/join框架支持任务拆分与合并，适用于并行计算。选择合适的并发模式需结合具体需求，例如简单共享变量可用原子类，读多写少用读写锁。避免死锁应破坏其四个必要条件，如避免嵌套锁、采用锁排序、设置超时机制或使用死锁检测工具。此外，静态分析、测试和性能分析工具也能辅助并发编程优化。掌握这些模式和工具能显著提升并发程序的稳定性和效率。

并发编程旨在提高程序效率，但稍有不慎，就会掉入各种陷阱。掌握一些核心模式，能帮助我们更好地应对并发挑战。

解决方案

并发编程的核心模式涵盖了线程管理、数据同步、任务调度等多个方面。以下列出十大核心模式，并提供代码模板，希望能帮助你快速上手：

1. 互斥锁（Mutex）

互斥锁用于保护临界区，确保同一时间只有一个线程可以访问共享资源。

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class MutexExample {

    private final Lock lock = new ReentrantLock();
    private int count = 0;

    public void increment() {
        lock.lock();
        try {
            count++;
        } finally {
            lock.unlock();
        }
    }

    public int getCount() {
        return count;
    }

    public static void main(String[] args) throws InterruptedException {
        MutexExample example = new MutexExample();

        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                example.increment();
            }
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                example.increment();
            }
        });

        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println("Count: " + example.getCount()); // Expected: 2000
    }
}

2. 读写锁（ReadWriteLock）

读写锁允许多个线程同时读取共享资源，但只允许一个线程写入。

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockExample {

    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private int data = 0;

    public int readData() {
        rwLock.readLock().lock();
        try {
            return data;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public void writeData(int newData) {
        rwLock.writeLock().lock();
        try {
            data = newData;
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    public static void main(String[] args) {
        ReadWriteLockExample example = new ReadWriteLockExample();

        // Writer thread
        new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                example.writeData(i);
                System.out.println("Writer: Wrote " + i);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // Reader threads
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                for (int j = 0; j < 10; j++) {
                    int value = example.readData();
                    System.out.println("Reader " + Thread.currentThread().getName() + ": Read " + value);
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "Reader-" + i).start();
        }
    }
}

3. 信号量（Semaphore）

信号量用于控制同时访问某个资源的线程数量。

import java.util.concurrent.Semaphore;

public class SemaphoreExample {

    private static final int MAX_PERMITS = 3;
    private final Semaphore semaphore = new Semaphore(MAX_PERMITS);

    public void accessResource(int threadId) {
        try {
            semaphore.acquire();
            System.out.println("Thread " + threadId + " acquired permit.");
            // Simulate resource access
            Thread.sleep((long) (Math.random() * 1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("Thread " + threadId + " releasing permit.");
            semaphore.release();
        }
    }

    public static void main(String[] args) {
        SemaphoreExample example = new SemaphoreExample();

        for (int i = 0; i < 5; i++) {
            final int threadId = i;
            new Thread(() -> example.accessResource(threadId)).start();
        }
    }
}

4. CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作。

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        for (int i = 0; i < 3; i++) {
            final int taskNumber = i;
            new Thread(() -> {
                try {
                    System.out.println("Task " + taskNumber + " is running...");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("Task " + taskNumber + " completed.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown();
                }
            }).start();
        }

        latch.await(); // Wait for all tasks to complete
        System.out.println("All tasks completed. Main thread continues.");
    }
}

5. CyclicBarrier

CyclicBarrier允许一组线程互相等待，直到所有线程都到达某个屏障点。

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {

    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All threads reached barrier!"));

        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            new Thread(() -> {
                try {
                    System.out.println("Thread " + threadId + " is working...");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("Thread " + threadId + " reached the barrier.");
                    barrier.await(); // Wait for other threads
                    System.out.println("Thread " + threadId + " continues after barrier.");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

6. Future/Callable

Future/Callable 允许异步执行任务并获取结果。

佳蓝在线销售系统(创业版) 佳蓝在线销售

1、对ASP内核代码进行DLL封装，从而大大提高了用户的访问速度和安全性；2、采用后台生成HTML网页的格式，使程序访问速度得到进一步的提升；3、用户可发展下级会员并在下级购买商品时获得差额利润；4、全新模板选择功能；5、后台增加磁盘绑定功能；6、后台增加库存查询功能；7、后台增加财务统计功能；8、后台面值类型批量设定；9、后台财务曲线报表显示；10、完善订单功能；11、对所有传输的字符串进行安全

下载

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class FutureCallableExample {

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        Callable task = () -> {
            System.out.println("Task is running...");
            Thread.sleep(1000);
            return "Task completed!";
        };

        Future future = executor.submit(task);

        System.out.println("Waiting for task to complete...");
        String result = future.get(); // Blocks until the task is done
        System.out.println("Result: " + result);

        executor.shutdown();
    }
}

7. 线程池（ThreadPool）

线程池用于管理和复用线程，减少线程创建和销毁的开销。

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++) {
            final int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Task " + taskNumber + " is running in thread: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Task " + taskNumber + " completed.");
            });
        }

        executor.shutdown();
    }
}

8. BlockingQueue

BlockingQueue 是一种线程安全的队列，可以在多个线程之间传递数据。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueExample {

    public static void main(String[] args) {
        BlockingQueue queue = new LinkedBlockingQueue<>(10);

        // Producer thread
        new Thread(() -> {
            try {
                for (int i = 0; i < 15; i++) {
                    queue.put(i); // Blocks if queue is full
                    System.out.println("Produced: " + i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // Consumer thread
        new Thread(() -> {
            try {
                while (true) {
                    Integer value = queue.take(); // Blocks if queue is empty
                    System.out.println("Consumed: " + value);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

9. 原子变量（Atomic Variables）

原子变量提供线程安全的原子操作，避免使用锁。

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerExample {

    private AtomicInteger count = new AtomicInteger(0);

    public void increment() {
        count.incrementAndGet();
    }

    public int getCount() {
        return count.get();
    }

    public static void main(String[] args) throws InterruptedException {
        AtomicIntegerExample example = new AtomicIntegerExample();

        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                example.increment();
            }
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                example.increment();
            }
        });

        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println("Count: " + example.getCount()); // Expected: 2000
    }
}

10. Fork/Join

Fork/Join 框架用于将大任务分解成小任务并行执行，然后合并结果。

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class ForkJoinExample {

    static class SumTask extends RecursiveTask {
        private static final int THRESHOLD = 100;
        private final long[] array;
        private final int start;
        private final int end;

        public SumTask(long[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Long compute() {
            if (end - start <= THRESHOLD) {
                long sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                return sum;
            } else {
                int middle = (start + end) / 2;
                SumTask leftTask = new SumTask(array, start, middle);
                SumTask rightTask = new SumTask(array, middle, end);

                leftTask.fork();
                rightTask.fork();

                return leftTask.join() + rightTask.join();
            }
        }
    }

    public static void main(String[] args) {
        long[] array = new long[1000];
        for (int i = 0; i < array.length; i++) {
            array[i] = i + 1;
        }

        ForkJoinPool pool = new ForkJoinPool();
        SumTask task = new SumTask(array, 0, array.length);

        long result = pool.invoke(task);
        System.out.println("Sum: " + result); // Expected: 500500
    }
}

如何选择合适的并发模式？

选择并发模式不能一概而论，需要根据具体场景来判断。比如，如果只需要保护一个简单的共享变量，原子变量可能就足够了，没必要动用互斥锁。如果读多写少，读写锁会更高效。

并发编程中常见的死锁问题如何避免？

死锁是并发编程中的一大难题。避免死锁的关键在于破坏死锁产生的四个必要条件：互斥、占有且等待、不可剥夺、环路等待。常用的方法包括：

• 避免嵌套锁： 尽量减少锁的持有时间，避免在一个锁的范围内获取另一个锁。
• 锁排序： 如果需要获取多个锁，按照固定的顺序获取，避免环路等待。
• 使用超时机制： 在获取锁时设置超时时间，如果超时未获取到锁，则释放已持有的锁。
• 死锁检测： 定期检测系统中是否存在死锁，如果发现死锁，则采取措施解除死锁。

除了代码模板，还有哪些工具可以辅助并发编程？

除了代码模板，还有一些工具可以辅助并发编程，例如：

• 静态分析工具： 可以帮助检查代码中的潜在并发问题，例如死锁、竞争条件等。
• 并发测试工具： 可以模拟高并发场景，帮助发现并发代码中的性能瓶颈和错误。
• 性能分析工具： 可以帮助分析并发程序的性能，找出性能瓶颈。

掌握这些核心模式和工具，能让你在并发编程的道路上走得更稳、更远。