Java 并发新特性从入门到精通实战教程详解

原创

啦啦啦191

发布于 2025-07-23 12:13:41

1970

文章被收录于专栏：Java开发Java开发

Java并发新特性与实战教程

随着Java版本的不断更新，并发编程领域引入了许多新特性和改进。本文将结合Java 8及后续版本的新特性，深入探讨并发编程的实战技巧，并通过具体案例展示如何利用这些新技术解决实际问题。

一、CompletableFuture：异步编程的革命

技术背景

Java 8引入的CompletableFuture彻底改变了异步编程的方式，它实现了Future和CompletionStage接口，支持链式调用和组合操作，避免了传统回调地狱的问题。

实操案例：电商订单处理系统

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class OrderProcessingSystem {
    private final ExecutorService executor = Executors.newFixedThreadPool(10);

    // 1. 校验订单信息
    public CompletableFuture<Order> validateOrder(Order order) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("校验订单: " + order.getId());
            // 模拟校验逻辑
            if (order.getAmount() <= 0) {
                throw new IllegalArgumentException("订单金额必须大于0");
            }
            return order;
        }, executor);
    }

    // 2. 扣减库存
    public CompletableFuture<Order> deductInventory(Order order) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("扣减库存: " + order.getProductId());
            // 模拟库存扣减
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return order;
        }, executor);
    }

    // 3. 支付处理
    public CompletableFuture<Order> processPayment(Order order) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("处理支付: " + order.getPaymentMethod());
            // 模拟支付处理
            try {
                Thread.sleep(800);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            order.setStatus(OrderStatus.PAID);
            return order;
        }, executor);
    }

    // 4. 发送通知
    public CompletableFuture<Void> sendNotification(Order order) {
        return CompletableFuture.runAsync(() -> {
            System.out.println("发送通知: " + order.getId());
            // 模拟通知发送
        }, executor);
    }

    // 组合所有操作
    public void processOrder(Order order) {
        validateOrder(order)
            .thenCompose(this::deductInventory)
            .thenCompose(this::processPayment)
            .thenAcceptAsync(this::sendNotification, executor)
            .exceptionally(ex -> {
                System.err.println("订单处理失败: " + ex.getMessage());
                return null;
            });
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        OrderProcessingSystem system = new OrderProcessingSystem();
        Order order = new Order(1L, "P001", 99.99, "ALIPAY");
        
        system.processOrder(order);
        
        // 主线程等待一段时间，确保异步任务完成
        Thread.sleep(2000);
        system.executor.shutdown();
    }
}

class Order {
    private Long id;
    private String productId;
    private double amount;
    private String paymentMethod;
    private OrderStatus status;

    // 构造方法、getter和setter略
}

enum OrderStatus {
    CREATED, PAID, SHIPPED, COMPLETED
}

技术要点

链式调用：通过thenCompose、thenAcceptAsync等方法实现异步操作的流水线处理。
异常处理：使用exceptionally方法捕获并处理整个流程中的异常。
自定义线程池：避免使用默认的ForkJoinPool，根据业务需求配置线程池大小。

二、StampedLock：读写锁的进化版

技术背景

Java 8引入的StampedLock是一种更高效的读写锁实现，支持乐观读模式，在读多写少的场景下性能显著提升。

实操案例：缓存系统

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.StampedLock;

public class CacheSystem<K, V> {
    private final Map<K, V> cache = new HashMap<>();
    private final StampedLock lock = new StampedLock();

    // 读操作：使用乐观读锁
    public V get(K key) {
        long stamp = lock.tryOptimisticRead();
        V value = cache.get(key);
        
        // 验证戳记有效性
        if (!lock.validate(stamp)) {
            // 升级为悲观读锁
            stamp = lock.readLock();
            try {
                value = cache.get(key);
            } finally {
                lock.unlockRead(stamp);
            }
        }
        return value;
    }

    // 写操作：使用写锁
    public void put(K key, V value) {
        long stamp = lock.writeLock();
        try {
            cache.put(key, value);
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    // 读改写操作：使用条件写锁
    public void updateIfExists(K key, V newValue) {
        long stamp = lock.readLock();
        try {
            if (!cache.containsKey(key)) {
                return;
            }
            
            // 升级为写锁
            long writeStamp = lock.tryConvertToWriteLock(stamp);
            if (writeStamp != 0) {
                // 升级成功
                stamp = writeStamp;
                cache.put(key, newValue);
            } else {
                // 升级失败，释放读锁，获取写锁
                lock.unlockRead(stamp);
                stamp = lock.writeLock();
                cache.put(key, newValue);
            }
        } finally {
            lock.unlock(stamp);
        }
    }
}

技术要点

乐观读锁：在读取频繁的场景下，通过tryOptimisticRead()避免阻塞写操作。
锁升级：通过tryConvertToWriteLock()方法实现锁的升级，减少锁的获取和释放开销。
条件写锁：在执行写操作前先检查条件，避免不必要的锁竞争。

三、Flow API：响应式流处理

技术背景

Java 9引入的Flow API（JEP 266）实现了响应式流规范（Reactive Streams），提供了非阻塞背压的异步流处理能力。

实操案例：实时数据流处理

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

// 1. 定义数据发布者
public class DataPublisher extends SubmissionPublisher<String> {
    public DataPublisher() {
        super();
    }

    public void publishData(String data) {
        submit(data);
    }
}

// 2. 定义数据处理器（中间操作）
public class DataProcessor implements Flow.Processor<String, String> {
    private Flow.Subscription subscription;
    private final SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

    @Override
    public void subscribe(Flow.Subscriber<? super String> subscriber) {
        publisher.subscribe(subscriber);
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1); // 请求第一个数据
    }

    @Override
    public void onNext(String item) {
        // 处理数据：转换为大写
        String processedData = item.toUpperCase();
        publisher.submit(processedData);
        subscription.request(1); // 请求下一个数据
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
        publisher.closeExceptionally(throwable);
    }

    @Override
    public void onComplete() {
        publisher.close();
    }
}

// 3. 定义数据订阅者
public class DataSubscriber implements Flow.Subscriber<String> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1); // 请求第一个数据
    }

    @Override
    public void onNext(String item) {
        System.out.println("处理数据: " + item);
        subscription.request(1); // 请求下一个数据
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("数据处理完成");
    }
}

// 4. 主程序：组装流处理管道
public class ReactiveStreamDemo {
    public static void main(String[] args) throws InterruptedException {
        try (DataPublisher publisher = new DataPublisher();
             DataProcessor processor = new DataProcessor()) {
            
            DataSubscriber subscriber = new DataSubscriber();
            
            // 组装流管道：发布者 -> 处理器 -> 订阅者
            publisher.subscribe(processor);
            processor.subscribe(subscriber);
            
            // 发布数据
            publisher.publishData("hello");
            publisher.publishData("world");
            publisher.publishData("java");
            
            // 等待所有数据处理完成
            Thread.sleep(1000);
        }
    }
}

技术要点

背压机制：通过request(n)方法实现消费者对生产者的流量控制。
处理器模式：使用Processor实现中间转换操作，构建复杂的流处理管道。
资源管理：使用try-with-resources确保Publisher正确关闭，避免资源泄漏。

四、VarHandle：内存访问的新方式

技术背景

Java 9引入的VarHandle提供了一种更高效、更灵活的内存访问机制，替代了传统的Unsafe类和Atomic类。

实操案例：高性能计数器

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;

public class HighPerformanceCounter {
    private static final VarHandle COUNTER;
    
    static {
        try {
            COUNTER = MethodHandles.lookup().findVarHandle(
                HighPerformanceCounter.class, 
                "counter", 
                long.class
            );
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new ExceptionInInitializerError(e);
        }
    }
    
    private volatile long counter = 0;
    
    // 原子递增
    public long increment() {
        return (long) COUNTER.getAndAdd(this, 1L);
    }
    
    // 获取当前值
    public long get() {
        return (long) COUNTER.get(this);
    }
    
    // 原子更新
    public boolean compareAndSet(long expected, long newValue) {
        return COUNTER.compareAndSet(this, expected, newValue);
    }
}

技术要点

直接内存访问：通过VarHandle直接操作内存，避免了反射的开销。
原子操作：支持getAndAdd、compareAndSet等原子操作，替代AtomicLong。
泛型支持：VarHandle是类型安全的，比Unsafe更可靠。

五、结构化并发：Java 19+ 的新特性

技术背景

Java 19引入的结构化并发（JEP 428）简化了多任务协作的管理，将多个相关任务视为一个工作单元，提高了可靠性和可观测性。

实操案例：用户资料聚合服务

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.StructuredTaskScope;

public class UserProfileService {
    private final ExecutorService executor = Executors.newFixedThreadPool(4);

    public UserProfile fetchUserProfile(String userId) throws InterruptedException {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            // 并行获取用户信息、订单信息和推荐商品
            var userInfoTask = scope.fork(() -> fetchUserInfo(userId));
            var orderTask = scope.fork(() -> fetchOrders(userId));
            var recommendationTask = scope.fork(() -> fetchRecommendations(userId));

            scope.join();           // 等待所有任务完成或任一任务失败
            scope.throwIfFailed();  // 如果有任务失败，抛出异常

            // 合并结果
            return new UserProfile(
                userInfoTask.get(),
                orderTask.get(),
                recommendationTask.get()
            );
        }
    }

    private UserInfo fetchUserInfo(String userId) {
        // 模拟从数据库获取用户信息
        return new UserInfo(userId, "张三", 30);
    }

    private Order[] fetchOrders(String userId) {
        // 模拟从订单服务获取订单列表
        return new Order[]{
            new Order("ORD123", userId, 299.0),
            new Order("ORD456", userId, 199.0)
        };
    }

    private Product[] fetchRecommendations(String userId) {
        // 模拟从推荐系统获取推荐商品
        return new Product[]{
            new Product("PRD001", "手机"),
            new Product("PRD002", "耳机")
        };
    }

    public static void main(String[] args) {
        UserProfileService service = new UserProfileService();
        try {
            UserProfile profile = service.fetchUserProfile("U12345");
            System.out.println(profile);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }
}

// 数据模型类略

技术要点