Jeskson-微服务分布式

Java内存模型与并发编程

# Java内存模型与并发编程

# 1. Java内存模型概述

# 1.1 什么是Java内存模型

Java内存模型（Java Memory Model，JMM）是Java虚拟机规范中定义的一个抽象概念，它描述了Java程序中各种变量（线程共享变量）的访问规则，以及在JVM中将变量存储到内存和从内存中读取变量这样的底层细节。

# 1.2 主内存与工作内存

public class MemoryModelDemo {
    private static int sharedVariable = 0;
    private static boolean flag = false;
    
    public static void demonstrateMemoryModel() {
        // 线程1：写入数据
        Thread writer = new Thread(() -> {
            System.out.println("Writer: 开始写入数据");
            sharedVariable = 42;
            flag = true;
            System.out.println("Writer: 数据写入完成");
        }, "Writer-Thread");
        
        // 线程2：读取数据
        Thread reader = new Thread(() -> {
            System.out.println("Reader: 开始读取数据");
            while (!flag) {
                // 等待flag变为true
                Thread.yield();
            }
            System.out.println("Reader: 读取到的值 = " + sharedVariable);
        }, "Reader-Thread");
        
        reader.start();
        try {
            Thread.sleep(100); // 确保reader先启动
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        writer.start();
        
        try {
            writer.join();
            reader.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        demonstrateMemoryModel();
    }
}

# 2. 内存间交互操作

# 2.1 八种基本操作

Java内存模型定义了8种操作来完成主内存与工作内存之间的交互：

  1. lock（锁定）：作用于主内存的变量
  2. unlock（解锁）：作用于主内存的变量
  3. read（读取）：作用于主内存的变量
  4. load（载入）：作用于工作内存的变量
  5. use（使用）：作用于工作内存的变量
  6. assign（赋值）：作用于工作内存的变量
  7. store（存储）：作用于工作内存的变量
  8. write（写入）：作用于主内存的变量
public class MemoryOperationsDemo {
    private static volatile int volatileVar = 0;
    private static int normalVar = 0;
    
    public static void demonstrateMemoryOperations() {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                // 对volatile变量的操作会立即同步到主内存
                volatileVar = i;
                normalVar = i;
                
                System.out.println("Thread1: volatileVar = " + volatileVar + 
                                 ", normalVar = " + normalVar);
                
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        
        Thread thread2 = new Thread(() -> {
            int lastVolatileValue = -1;
            int lastNormalValue = -1;
            
            for (int i = 0; i < 10; i++) {
                int currentVolatile = volatileVar;
                int currentNormal = normalVar;
                
                if (currentVolatile != lastVolatileValue || 
                    currentNormal != lastNormalValue) {
                    System.out.println("Thread2: 读取到 volatileVar = " + currentVolatile + 
                                     ", normalVar = " + currentNormal);
                    lastVolatileValue = currentVolatile;
                    lastNormalValue = currentNormal;
                }
                
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        
        thread1.start();
        thread2.start();
        
        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        demonstrateMemoryOperations();
    }
}

# 3. volatile关键字

# 3.1 volatile的特性

public class VolatileDemo {
    private static volatile boolean stopFlag = false;
    private static boolean normalFlag = false;
    
    // 演示volatile的可见性
    public static void demonstrateVisibility() {
        System.out.println("=== 演示volatile可见性 ===");
        
        // 使用volatile变量的线程
        Thread volatileThread = new Thread(() -> {
            int count = 0;
            while (!stopFlag) {
                count++;
                if (count % 100000000 == 0) {
                    System.out.println("Volatile线程运行中... count = " + count);
                }
            }
            System.out.println("Volatile线程结束，count = " + count);
        }, "Volatile-Thread");
        
        // 使用普通变量的线程
        Thread normalThread = new Thread(() -> {
            int count = 0;
            while (!normalFlag) {
                count++;
                if (count % 100000000 == 0) {
                    System.out.println("Normal线程运行中... count = " + count);
                }
            }
            System.out.println("Normal线程结束，count = " + count);
        }, "Normal-Thread");
        
        volatileThread.start();
        normalThread.start();
        
        try {
            Thread.sleep(1000);
            System.out.println("主线程：设置停止标志");
            stopFlag = true;
            normalFlag = true;
            
            // 等待线程结束
            volatileThread.join(2000);
            if (normalThread.isAlive()) {
                System.out.println("Normal线程可能因为可见性问题未能及时停止");
                normalThread.interrupt();
            }
            normalThread.join(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 演示volatile禁止指令重排序
    private static volatile boolean initialized = false;
    private static int value = 0;
    
    public static void demonstrateOrdering() {
        System.out.println("\n=== 演示volatile禁止重排序 ===");
        
        Thread initThread = new Thread(() -> {
            System.out.println("初始化线程：开始初始化");
            value = 42; // 1. 设置值
            initialized = true; // 2. 设置初始化标志（volatile写）
            System.out.println("初始化线程：初始化完成");
        }, "Init-Thread");
        
        Thread useThread = new Thread(() -> {
            System.out.println("使用线程：等待初始化完成");
            while (!initialized) { // volatile读
                Thread.yield();
            }
            // 由于volatile的happens-before规则，这里一定能看到value=42
            System.out.println("使用线程：读取到value = " + value);
        }, "Use-Thread");
        
        useThread.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        initThread.start();
        
        try {
            initThread.join();
            useThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        demonstrateVisibility();
        demonstrateOrdering();
    }
}

# 3.2 volatile的使用场景

import java.util.concurrent.atomic.AtomicInteger;

public class VolatileUseCases {
    
    // 场景1：状态标志
    private static volatile boolean shutdown = false;
    
    public static void statusFlagExample() {
        System.out.println("=== 状态标志示例 ===");
        
        Thread worker = new Thread(() -> {
            while (!shutdown) {
                // 执行工作
                System.out.println("工作线程正在运行...");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
            }
            System.out.println("工作线程已停止");
        });
        
        worker.start();
        
        try {
            Thread.sleep(2000);
            System.out.println("主线程：发送停止信号");
            shutdown = true;
            worker.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 场景2：一次性安全发布
    private static volatile Resource resource;
    
    static class Resource {
        private final String data;
        
        public Resource(String data) {
            this.data = data;
        }
        
        public String getData() {
            return data;
        }
    }
    
    public static void safePublicationExample() {
        System.out.println("\n=== 安全发布示例 ===");
        
        Thread publisher = new Thread(() -> {
            System.out.println("发布者：创建资源");
            Resource newResource = new Resource("重要数据");
            resource = newResource; // volatile写，确保安全发布
            System.out.println("发布者：资源发布完成");
        });
        
        Thread consumer = new Thread(() -> {
            System.out.println("消费者：等待资源");
            while (resource == null) {
                Thread.yield();
            }
            System.out.println("消费者：获取到资源 - " + resource.getData());
        });
        
        consumer.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        publisher.start();
        
        try {
            publisher.join();
            consumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 场景3：独立观察（注意：volatile不保证原子性）
    private static volatile int counter = 0;
    private static AtomicInteger atomicCounter = new AtomicInteger(0);
    
    public static void atomicityExample() {
        System.out.println("\n=== 原子性示例（volatile vs atomic） ===");
        
        int threadCount = 5;
        int incrementsPerThread = 1000;
        Thread[] threads = new Thread[threadCount];
        
        // 使用volatile变量（不保证原子性）
        for (int i = 0; i < threadCount; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < incrementsPerThread; j++) {
                    counter++; // 非原子操作
                    atomicCounter.incrementAndGet(); // 原子操作
                }
            });
        }
        
        for (Thread thread : threads) {
            thread.start();
        }
        
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        int expected = threadCount * incrementsPerThread;
        System.out.println("期望值: " + expected);
        System.out.println("volatile counter: " + counter + 
                         " (可能小于期望值，因为++不是原子操作)");
        System.out.println("atomic counter: " + atomicCounter.get() + 
                         " (应该等于期望值)");
    }
    
    public static void main(String[] args) {
        statusFlagExample();
        safePublicationExample();
        atomicityExample();
    }
}

# 4. happens-before规则

# 4.1 happens-before关系

import java.util.concurrent.locks.ReentrantLock;

public class HappensBeforeDemo {
    private static int x = 0;
    private static int y = 0;
    private static volatile boolean flag = false;
    private static final ReentrantLock lock = new ReentrantLock();
    
    // 规则1：程序次序规则
    public static void programOrderRule() {
        System.out.println("=== 程序次序规则 ===");
        
        Thread thread = new Thread(() -> {
            x = 1; // 操作A
            y = 2; // 操作B
            // 根据程序次序规则，A happens-before B
            System.out.println("x = " + x + ", y = " + y);
        });
        
        thread.start();
        try {
            thread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 规则2：volatile变量规则
    public static void volatileVariableRule() {
        System.out.println("\n=== volatile变量规则 ===");
        
        Thread writer = new Thread(() -> {
            x = 42;
            flag = true; // volatile写
            System.out.println("Writer: 设置x=42, flag=true");
        });
        
        Thread reader = new Thread(() -> {
            if (flag) { // volatile读
                // 由于volatile规则，writer中的x=42 happens-before 这里的读取
                System.out.println("Reader: flag为true，x = " + x);
            }
        });
        
        writer.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        reader.start();
        
        try {
            writer.join();
            reader.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 规则3：锁规则
    public static void lockRule() {
        System.out.println("\n=== 锁规则 ===");
        
        Thread thread1 = new Thread(() -> {
            lock.lock();
            try {
                x = 100;
                System.out.println("Thread1: 在锁内设置x=100");
            } finally {
                lock.unlock(); // 解锁操作
            }
        });
        
        Thread thread2 = new Thread(() -> {
            lock.lock(); // 加锁操作
            try {
                // 由于锁规则，thread1的解锁 happens-before thread2的加锁
                // 因此thread2能看到thread1对x的修改
                System.out.println("Thread2: 在锁内读取x = " + x);
            } finally {
                lock.unlock();
            }
        });
        
        thread1.start();
        try {
            thread1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        thread2.start();
        try {
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 规则4：线程启动规则
    public static void threadStartRule() {
        System.out.println("\n=== 线程启动规则 ===");
        
        x = 200; // 主线程中的操作
        
        Thread childThread = new Thread(() -> {
            // 由于线程启动规则，主线程中start()之前的操作 happens-before 子线程中的操作
            System.out.println("子线程: 读取到x = " + x);
        });
        
        childThread.start(); // 启动子线程
        
        try {
            childThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 规则5：线程终止规则
    public static void threadTerminationRule() {
        System.out.println("\n=== 线程终止规则 ===");
        
        Thread childThread = new Thread(() -> {
            y = 300;
            System.out.println("子线程: 设置y=300");
        });
        
        childThread.start();
        
        try {
            childThread.join(); // 等待子线程结束
            // 由于线程终止规则，子线程中的操作 happens-before join()的返回
            System.out.println("主线程: 子线程结束后读取y = " + y);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        programOrderRule();
        volatileVariableRule();
        lockRule();
        threadStartRule();
        threadTerminationRule();
    }
}

# 5. 内存屏障

# 5.1 内存屏障类型

import sun.misc.Unsafe;
import java.lang.reflect.Field;

public class MemoryBarrierDemo {
    private static Unsafe unsafe;
    
    static {
        try {
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            unsafe = (Unsafe) field.get(null);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static volatile int volatileVar = 0;
    private static int normalVar1 = 0;
    private static int normalVar2 = 0;
    
    // 演示内存屏障的作用
    public static void demonstrateMemoryBarrier() {
        System.out.println("=== 内存屏障演示 ===");
        
        Thread writer = new Thread(() -> {
            // 写入普通变量
            normalVar1 = 1;
            normalVar2 = 2;
            
            // 插入StoreStore屏障（通过volatile写实现）
            volatileVar = 3;
            
            System.out.println("Writer: 完成写入操作");
        });
        
        Thread reader = new Thread(() -> {
            // 读取volatile变量（插入LoadLoad和LoadStore屏障）
            int vol = volatileVar;
            
            if (vol == 3) {
                // 由于内存屏障，这里能看到之前的写入
                System.out.println("Reader: volatileVar=" + vol + 
                                 ", normalVar1=" + normalVar1 + 
                                 ", normalVar2=" + normalVar2);
            }
        });
        
        writer.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        reader.start();
        
        try {
            writer.join();
            reader.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 使用Unsafe直接操作内存屏障（仅用于演示，实际开发不推荐）
    public static void unsafeMemoryBarrier() {
        if (unsafe == null) {
            System.out.println("Unsafe不可用，跳过演示");
            return;
        }
        
        System.out.println("\n=== Unsafe内存屏障演示 ===");
        
        Thread thread = new Thread(() -> {
            normalVar1 = 10;
            
            // 插入StoreStore屏障
            unsafe.storeFence();
            
            normalVar2 = 20;
            
            // 插入StoreLoad屏障
            unsafe.fullFence();
            
            System.out.println("使用Unsafe插入内存屏障完成");
        });
        
        thread.start();
        try {
            thread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        demonstrateMemoryBarrier();
        unsafeMemoryBarrier();
    }
}

# 6. 双重检查锁定

# 6.1 单例模式的内存模型问题

public class DoubleCheckedLockingDemo {
    
    // 错误的双重检查锁定实现
    static class BrokenSingleton {
        private static BrokenSingleton instance;
        
        private BrokenSingleton() {
            // 模拟复杂的初始化过程
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        // 存在问题的实现
        public static BrokenSingleton getInstance() {
            if (instance == null) { // 第一次检查
                synchronized (BrokenSingleton.class) {
                    if (instance == null) { // 第二次检查
                        instance = new BrokenSingleton(); // 问题所在
                    }
                }
            }
            return instance;
        }
    }
    
    // 正确的双重检查锁定实现
    static class CorrectSingleton {
        private static volatile CorrectSingleton instance;
        
        private CorrectSingleton() {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        public static CorrectSingleton getInstance() {
            if (instance == null) { // 第一次检查
                synchronized (CorrectSingleton.class) {
                    if (instance == null) { // 第二次检查
                        instance = new CorrectSingleton(); // volatile确保正确发布
                    }
                }
            }
            return instance;
        }
    }
    
    // 基于类初始化的线程安全单例
    static class InitializationOnDemandSingleton {
        private InitializationOnDemandSingleton() {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        private static class InstanceHolder {
            static final InitializationOnDemandSingleton INSTANCE = 
                new InitializationOnDemandSingleton();
        }
        
        public static InitializationOnDemandSingleton getInstance() {
            return InstanceHolder.INSTANCE;
        }
    }
    
    public static void testSingletonImplementations() {
        System.out.println("=== 测试单例模式实现 ===");
        
        // 测试正确的双重检查锁定
        Thread[] threads = new Thread[10];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(() -> {
                CorrectSingleton singleton = CorrectSingleton.getInstance();
                System.out.println("CorrectSingleton实例: " + 
                                 System.identityHashCode(singleton));
            });
        }
        
        for (Thread thread : threads) {
            thread.start();
        }
        
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        System.out.println();
        
        // 测试基于类初始化的单例
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                InitializationOnDemandSingleton singleton = 
                    InitializationOnDemandSingleton.getInstance();
                System.out.println("InitializationOnDemandSingleton实例: " + 
                                 System.identityHashCode(singleton));
            }).start();
        }
    }
    
    public static void main(String[] args) {
        testSingletonImplementations();
    }
}

# 7. 总结

# 7.1 关键概念

  1. Java内存模型：定义了线程间如何通过内存进行交互的规范
  2. 主内存与工作内存：JMM的抽象概念，用于描述变量的存储和访问
  3. volatile关键字：保证可见性和有序性，但不保证原子性
  4. happens-before规则：定义了操作间的偏序关系
  5. 内存屏障：防止指令重排序的硬件或软件机制

# 7.2 最佳实践

  1. 正确使用volatile：用于状态标志、安全发布等场景
  2. 理解happens-before：确保线程间的可见性
  3. 避免双重检查锁定陷阱：使用volatile或其他线程安全方式
  4. 选择合适的同步机制：根据需求选择synchronized、volatile、Lock等
  5. 关注内存可见性：在多线程环境中确保数据的正确同步

# 7.3 常见问题

  1. 可见性问题：一个线程的修改对其他线程不可见
  2. 重排序问题：编译器或处理器的优化导致执行顺序改变
  3. 原子性问题：复合操作不是原子的
  4. 双重检查锁定问题：不正确的单例实现可能导致问题

理解Java内存模型是编写正确并发程序的基础，它帮助我们理解多线程程序的行为，避免常见的并发问题。

Java并发编程最佳实践