IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java线程协作案例

wen java案例 4

本文目录导读：

Java线程协作案例

  1. 生产者-消费者模式（经典案例）
  2. 使用并发工具类实现更优雅的方案
  3. CountDownLatch 协作案例
  4. CyclicBarrier 循环屏障案例
  5. Semaphore 信号量案例
  6. CompletableFuture 异步协作
  7. 实战：并发处理任务汇总

我来为你提供几个Java线程协作的经典案例,从简单到复杂逐步展开。

生产者-消费者模式（经典案例）

这是最经典的线程协作案例,使用wait()和notifyAll()实现。

import java.util.LinkedList;
import java.util.Queue;
public class ProducerConsumerExample {
    // 共享缓冲区
    private static final Queue<Integer> BUFFER = new LinkedList<>();
    private static final int CAPACITY = 5;
    // 生产者
    static class Producer implements Runnable {
        private final String name;
        public Producer(String name) {
            this.name = name;
        }
        @Override
        public void run() {
            int value = 0;
            while (true) {
                synchronized (BUFFER) {
                    // 缓冲区满时等待
                    while (BUFFER.size() == CAPACITY) {
                        try {
                            System.out.println(name + " 等待，缓冲区已满...");
                            BUFFER.wait();
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            return;
                        }
                    }
                    // 生产数据
                    value++;
                    BUFFER.offer(value);
                    System.out.println(name + " 生产: " + value + "，队列大小: " + BUFFER.size());
                    // 唤醒其他线程
                    BUFFER.notifyAll();
                }
                // 模拟生产时间
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }
    // 消费者
    static class Consumer implements Runnable {
        private final String name;
        public Consumer(String name) {
            this.name = name;
        }
        @Override
        public void run() {
            while (true) {
                synchronized (BUFFER) {
                    // 缓冲区空时等待
                    while (BUFFER.isEmpty()) {
                        try {
                            System.out.println(name + " 等待，缓冲区为空...");
                            BUFFER.wait();
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            return;
                        }
                    }
                    // 消费数据
                    int value = BUFFER.poll();
                    System.out.println(name + " 消费: " + value + "，队列大小: " + BUFFER.size());
                    // 唤醒其他线程
                    BUFFER.notifyAll();
                }
                // 模拟消费时间
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }
    public static void main(String[] args) {
        // 创建线程
        Thread p1 = new Thread(new Producer("生产者1"));
        Thread p2 = new Thread(new Producer("生产者2"));
        Thread c1 = new Thread(new Consumer("消费者1"));
        Thread c2 = new Thread(new Consumer("消费者2"));
        // 启动线程
        p1.start();
        p2.start();
        c1.start();
        c2.start();
    }
}

使用并发工具类实现更优雅的方案

使用BlockingQueue让代码更简洁、更安全：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class BlockingQueueExample {
    private static final BlockingQueue<Integer> QUEUE = new ArrayBlockingQueue<>(5);
    static class Producer implements Runnable {
        @Override
        public void run() {
            int value = 0;
            while (true) {
                try {
                    value++;
                    QUEUE.put(value); // 自动阻塞
                    System.out.println("生产: " + value);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }
    static class Consumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    Integer value = QUEUE.take(); // 自动阻塞
                    System.out.println("消费: " + value);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }
    public static void main(String[] args) {
        new Thread(new Producer(), "生产者").start();
        new Thread(new Consumer(), "消费者").start();
    }
}

CountDownLatch 协作案例

多个线程等待某个条件满足后再继续执行：

import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        int taskCount = 3;
        CountDownLatch latch = new CountDownLatch(taskCount);
        // 创建三个任务线程
        for (int i = 1; i <= taskCount; i++) {
            final int taskId = i;
            new Thread(() -> {
                try {
                    System.out.println("任务 " + taskId + " 开始执行");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("任务 " + taskId + " 完成");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown(); // 任务完成，计数器减1
                }
            }).start();
        }
        // 主线程等待所有任务完成
        System.out.println("主线程等待所有任务完成...");
        latch.await();
        System.out.println("所有任务已完成，继续执行主线程");
    }
}

CyclicBarrier 循环屏障案例

让一组线程互相等待,达到某个共同点后再一起继续执行：

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
    public static void main(String[] args) {
        int playerCount = 4;
        CyclicBarrier barrier = new CyclicBarrier(playerCount, () -> {
            System.out.println("所有玩家已就绪，游戏开始！");
        });
        // 创建玩家线程
        for (int i = 1; i <= playerCount; i++) {
            final int playerId = i;
            new Thread(() -> {
                try {
                    System.out.println("玩家 " + playerId + " 正在加载资源...");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("玩家 " + playerId + " 已就绪");
                    // 等待其他玩家
                    barrier.await();
                    // 游戏开始后的操作
                    System.out.println("玩家 " + playerId + " 进入游戏！");
                } catch (InterruptedException | BrokenBarrierException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }
    }
}

Semaphore 信号量案例

控制并发访问资源的线程数量：

import java.util.concurrent.Semaphore;
public class SemaphoreExample {
    // 图书馆只有3个座位
    private static final Semaphore SEATS = new Semaphore(3);
    static class Student implements Runnable {
        private final String name;
        public Student(String name) {
            this.name = name;
        }
        @Override
        public void run() {
            try {
                System.out.println(name + " 等待座位...");
                SEATS.acquire(); // 获取许可
                System.out.println(name + " 获得座位，开始学习");
                Thread.sleep((long) (Math.random() * 2000));
                System.out.println(name + " 学习完毕，离开座位");
                SEATS.release(); // 释放许可
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    public static void main(String[] args) {
        String[] students = {"小明", "小红", "小刚", "小丽", "小明2号", "小红2号"};
        for (String name : students) {
            new Thread(new Student(name)).start();
        }
    }
}

CompletableFuture 异步协作

现代Java中优雅的异步编程方式：

import java.util.concurrent.CompletableFuture;
public class CompletableFutureExample {
    public static void main(String[] args) {
        // 模拟订单流程：下单 → 支付 → 发货 → 完成
        CompletableFuture.supplyAsync(() -> {
            // 步骤1：下单
            System.out.println("1. 用户下单，订单号: " + Math.random());
            try { Thread.sleep(1000); } catch (InterruptedException e) {}
            return "订单已创建";
        }).thenApply(order -> {
            // 步骤2：支付
            System.out.println("2. 处理支付");
            try { Thread.sleep(1000); } catch (InterruptedException e) {}
            return order + "，支付成功";
        }).thenApply(payment -> {
            // 步骤3：发货
            System.out.println("3. 商品发货");
            try { Thread.sleep(1000); } catch (InterruptedException e) {}
            return payment + "，已发货";
        }).thenAccept(result -> {
            // 步骤4：完成
            System.out.println("4. 订单完成: " + result);
        });
        // 等待异步任务完成
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

实战：并发处理任务汇总

一个综合的实战案例,演示如何协作处理大量任务：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
public class TaskProcessingExample {
    static class TaskResult {
        private final int taskId;
        private final String data;
        public TaskResult(int taskId, String data) {
            this.taskId = taskId;
            this.data = data;
        }
        @Override
        public String toString() {
            return "任务" + taskId + ": " + data;
        }
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 模拟100个需要处理的任务
        List<Integer> tasks = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            tasks.add(i);
        }
        // 使用线程池并行处理
        ExecutorService executor = Executors.newFixedThreadPool(4);
        // 提交所有任务并收集结果
        List<CompletableFuture<TaskResult>> futures = tasks.stream()
            .map(taskId -> CompletableFuture.supplyAsync(() -> processTask(taskId), executor))
            .collect(Collectors.toList());
        // 等待所有任务完成
        CompletableFuture<Void> allDone = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        // 任务都完成后执行回调
        allDone.thenAccept(voidPtr -> {
            System.out.println("所有任务已完成，正在汇总结果...");
            // 收集结果
            List<TaskResult> results = futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
            // 打印结果
            results.forEach(System.out::println);
            System.out.println("总计处理任务数量: " + results.size());
        }).join();
        // 关闭线程池
        executor.shutdown();
    }
    private static TaskResult processTask(int taskId) {
        try {
            System.out.println("任务" + taskId + " 开始处理 (线程: " + 
                Thread.currentThread().getName() + ")");
            // 模拟任务处理时间
            Thread.sleep((long) (Math.random() * 1000));
            String data = "处理数据-" + taskId;
            System.out.println("任务" + taskId + " 处理完成");
            return new TaskResult(taskId, data);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return new TaskResult(taskId, "处理失败");
        }
    }
}
  1. synchronized + wait/notify：最基础的协作方式，需注意在循环中检查条件
  2. BlockingQueue：无需手动管理同步，自动阻塞/唤醒
  3. CountDownLatch：一次性事件，主线程等待所有子任务完成
  4. CyclicBarrier：循环使用，多个线程互相等待到共同点
  5. Semaphore：控制并发访问数量，类似流量控制
  6. CompletableFuture：现代异步编程，支持链式调用

选择建议：

  • 简单场景用 synchronized 或 BlockingQueue
  • 等待多任务完成用 CountDownLatch
  • 循环等待点协作用 CyclicBarrier
  • 控制并发数量用 Semaphore
  • 异步编程优先考虑 CompletableFuture

希望这些案例能帮助你理解Java线程协作的机制！

上一篇拒绝策略案例

下一篇Java线程池参数案例

相关文章

抱歉，评论功能暂时关闭!