Java并发编程——无锁(乐观锁)的方式实现并发

最新推荐文章于 2025-04-12 10:56:55 发布

程勇uestc

最新推荐文章于 2025-04-12 10:56:55 发布

阅读量874

点赞数

分类专栏： Java 文章标签：乐观锁 CAS 无锁不可变类设计享元模式

本文链接：https://blog.csdn.net/qq_40438165/article/details/115917656

版权

Java 专栏收录该内容

51 篇文章

订阅专栏

1、CAS

$\quad$ 使用CAS(compare and swap)来实现无锁时线程安全，因为CAS操作底层是原子的。其实CAS底层是lock cmpxchg指令，在单核和多核CPU下都能够保证比较-交换的原子性。CAS操作需要volatile的支持，需要保证变量的可见性，因此可以用AtomicInteger代替int，其内部使用了volatile修饰。
$\quad$ CAS效率比synchronized效率高，原因？无锁情况下，即使重试失败，线程始终在高速运行，没有停歇，而 synchronized 会让线程在没有获得锁的时候，发生上下文切换，进入阻塞。在线程数小于cpu核心数时，使用cas是非常合适的，单核情况下cas不如sync。
$\quad$ CAS特点：

结合 CAS 和 volatile 可以实现无锁并发，适用于线程数少、多核 CPU 的场景下
CAS 是基于乐观锁的思想：最乐观的估计，不怕别的线程来修改共享变量，就算改了也没关系，我吃亏点重试呗
synchronized 是基于悲观锁的思想：最悲观的估计，得防着其它线程来修改共享变量，我上了锁你们都别想改，我改完了解开锁，你们才有机会
CAS 体现的是无锁并发、无阻塞并发

public class TestAccount {
    public static void main(String[] args) {
        Account account1 = new AccountCas(100000);
        Account.demo(account1);

        Account account2 = new AccountUnsafe(100000);
        Account.demo(account2);
    }
}
class AccountCas implements Account{
    private AtomicInteger balance;

    public AccountCas(int balance) {
        this.balance = new AtomicInteger(balance);
    }

    @Override
    public Integer getBalance() {
        return balance.get();
    }

    @Override
    public void withdraw(Integer amount) {
        while(true){
            int prev = balance.get();  // 获取余额最新值
            int next = prev - 10;  // 要修改的余额
            // 真正修改
            if(balance.compareAndSet(prev, next)){
                break;
            }
        }
    }
}
class AccountUnsafe implements Account{
    private Integer balance;

    public AccountUnsafe(Integer balance) {
        this.balance = balance;
    }

    @Override
    public Integer getBalance() {
        synchronized (this) {
            return balance;
        }
    }

    @Override
    public void withdraw(Integer amount) {
        synchronized (this) {
            this.balance -= 10;
        }
    }
}

interface Account {
    // 获取余额
    Integer getBalance();

    // 取款
    void withdraw(Integer amount);

    /**
     * 方法内会启动 1000 个线程，每个线程做 -10 元 的操作
     * 如果初始余额为 10000 那么正确的结果应当是 0
     */
    static void demo(Account account) {
        List<Thread> ts = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            ts.add(new Thread(() -> {
                account.withdraw(10);
            }));
        }
        long start = System.nanoTime();
        ts.forEach(Thread::start);
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        long end = System.nanoTime();
        System.out.println(account.getBalance()
                + " cost: " + (end-start)/1000_000 + " ms");
    }
}

2、为并发准备的原子数据类型

$\quad$ JUC包下的原子数据类型能保证其操作是原子的，例如原子整数进行加法，能保证进行加法的几点指令的原子性。

1、原子整数 `AtomicInteger`

    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(0);
        System.out.println(i.get());  // 0
        i.compareAndSet(0, -1);  // 如果当前i是0，则用-1交换它；如果不是则不交换
        System.out.println(i);  // -1

        System.out.println(i.incrementAndGet());  // 0   ++ i
        System.out.println(i.decrementAndGet());  // -1  -- i
        System.out.println(i.addAndGet(6));  // 5  i += 6
        System.out.println(i.updateAndGet(value -> value * 10));  // 25 5*10=50
    }

$\quad$ 手动实现updateAndGet功能：

    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(5);
        while(true){
            int prev = i.get();
            int next = prev * 10;
            if(i.compareAndSet(prev, next)){
                break;
            }
        }
        System.out.println(i.get());  // 50
    }

$\quad$ 可以封装成函数：

    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(5);
        System.out.println(updateAndGet(i, p -> p * 10));
    }

    public static int updateAndGet(AtomicInteger i, IntUnaryOperator operator){
        while(true){
            int prev = i.get();
            int next = operator.applyAsInt(prev);
            if(i.compareAndSet(prev, next)){
                return next;
            }
        }
    }

2、原子引用

$\quad$ 需要保证原子性的操作不一定只有整数，可能是各种类，这个时候我们就需要使用原子引用了AtomicReference<BigDecimal>。

class DecimalAccountCas implements DecimalAccount {
    private AtomicReference<BigDecimal> balance;

    public DecimalAccountCas(BigDecimal balance) {
        this.balance = new AtomicReference<>(balance);
    }

    @Override
    public BigDecimal getBalance() {
        return balance.get();
    }

    @Override
    public void withdraw(BigDecimal amount) {
        while(true){
            BigDecimal prev = balance.get();
            BigDecimal next = prev.subtract(amount);
            if(balance.compareAndSet(prev, next)){
                break;
            }
        }
    }
}

$\quad$ CAS的ABA问题：

共享变量c初始值为A，被线程t1修改为B后又被线程t2修改为A，此时再进行CAS就会出现问题
AtomicReference感知不到共享变量变换多次回到开始值的情况
如果主线程希望只要有其它线程动过了共享变量，那么自己的 cas 就算失败，这时，仅比较值是不够的，需要再加一个版本号
AtomicStampedReference 可以给原子引用加上版本号，追踪原子引用整个的变化过程
但是有时候，并不关心引用变量更改了几次，只是单纯的关心是否更改过，所以就有了AtomicMarkableReference

static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);

public static void main(String[] args) throws InterruptedException {
    log.debug("main start...");
    // 获取值 A
    String prev = ref.getReference();
    // 获取版本号
    int stamp = ref.getStamp();
    log.debug("stamp {}", stamp);
    // 如果中间有其它线程干扰，发生了 ABA 现象
    other();
    sleep(1);
    // 尝试改为 C
    log.debug("change A->C {}", ref.compareAndSet(prev, "C", stamp, stamp + 1));
}

private static void other() {
    new Thread(() -> {
        log.debug("change A->B {}", ref.compareAndSet(ref.getReference(), "B", ref.getStamp(), ref.getStamp() + 1));
        log.debug("new stamp {}", ref.getStamp());
    }, "t1").start();
    sleep(0.5);
    new Thread(() -> {
        log.debug("change B->A {}", ref.compareAndSet(ref.getReference(), "A", ref.getStamp(), ref.getStamp() + 1));
        log.debug("new stamp {}", ref.getStamp());
    }, "t2").start();
}

3、原子数组

$\quad$ 原子引用只能保证对象引用不发生改变，假设对象引用内值发生了改变，原子引用是无法识别的。这时候，就可以使用原子数组了：

AtomicIntegerArray
AtomicLongArray
AtomicReferenceArray

public static void main(String[] args) {
    demo(
            ()->new int[10],  // 不是原子数组，线程不安全
            (array)->array.length,
            (array, index) -> array[index]++,
            array-> System.out.println(Arrays.toString(array))
    );

    demo(
            ()-> new AtomicIntegerArray(10),  // 原子数组，线程安全
            (array) -> array.length(),
            (array, index) -> array.getAndIncrement(index),
            array -> System.out.println(array)
    );
}

/**
 参数1，提供数组、可以是线程不安全数组或线程安全数组
 参数2，获取数组长度的方法
 参数3，自增方法，回传 array, index
 参数4，打印数组的方法
 */
// supplier 提供者 无中生有  ()->结果
// function 函数   一个参数一个结果   (参数)->结果  ,  BiFunction (参数1,参数2)->结果
// consumer 消费者 一个参数没结果  (参数)->void,      BiConsumer (参数1,参数2)->
private static <T> void demo(
        Supplier<T> arraySupplier,
        Function<T, Integer> lengthFun,
        BiConsumer<T, Integer> putConsumer,
        Consumer<T> printConsumer ) {
    List<Thread> ts = new ArrayList<>();
    T array = arraySupplier.get();
    int length = lengthFun.apply(array);
    for (int i = 0; i < length; i++) {
        // 每个线程对数组作 10000 次操作
        ts.add(new Thread(() -> {
            for (int j = 0; j < 10000; j++) {
                putConsumer.accept(array, j%length);
            }
        }));
    }

    ts.forEach(t -> t.start()); // 启动所有线程
    ts.forEach(t -> {
        try {
            t.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });     // 等所有线程结束
    printConsumer.accept(array);
}

[6269, 6328, 6410, 6347, 6282, 6291, 6325, 6323, 6296, 6296]
[10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000]

4、字段更新器

$\quad$ 利用字段更新器，可以针对对象的某个域(Field)进行原子操作，只能配合volatile修饰的字段使用。

AtomicReferenceFieldApdater
AtomicIntegerFieldApdater
AtomicLongFieldApdater

public class Test40 {

    public static void main(String[] args) {
        Student stu = new Student();

        AtomicReferenceFieldUpdater updater =
                AtomicReferenceFieldUpdater.newUpdater(Student.class, String.class, "name");

        System.out.println(updater.compareAndSet(stu, null, "cy"));
        System.out.println(stu);
    }
}

class Student {
    volatile String name;  // 必须使用volatile修饰保证共享变量可见性

    @Override
    public String toString() {
        return "Student{" +
                "name='" + name + '\'' +
                '}';
    }
}

Student{name='cy'}

5、原子累加器

$\quad$ AutomicLong之类的也可以完成累加，但是效率不高，使用原子累加器LongAdder速度会更快，如下实例：

public class Test41 {
    public static void main(String[] args) {
        for (int i = 0; i < 5; i++) {
            demo(
                    () -> new AtomicLong(0),
                    (adder) -> adder.getAndIncrement()
            );
        }

        for (int i = 0; i < 5; i++) {
            demo(
                    () -> new LongAdder(),
                    adder -> adder.increment()
            );
        }
    }

    /*
    () -> 结果    提供累加器对象
    (参数) ->     执行累加操作
     */
    private static <T> void demo(Supplier<T> adderSupplier, Consumer<T> action) {
        T adder = adderSupplier.get();
        List<Thread> ts = new ArrayList<>();
        // 4 个线程，每人累加 50 万
        for (int i = 0; i < 4; i++) {
            ts.add(new Thread(() -> {
                for (int j = 0; j < 5000000; j++) {
                    action.accept(adder);
                }
            }));
        }
        long start = System.nanoTime();
        ts.forEach(t -> t.start());
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        long end = System.nanoTime();
        System.out.println(adder + " cost:" + (end - start) / 1000_000);
    }
}

20000000 cost:406
20000000 cost:379
20000000 cost:338
20000000 cost:348
20000000 cost:374
20000000 cost:54
20000000 cost:41
20000000 cost:42
20000000 cost:43
20000000 cost:40

$\quad$ 性能提升的原因很简单，就是在有竞争时，设置多个累加单元，Therad-0 累加 Cell[0]，而 Thread-1 累加Cell[1]… 最后将结果汇总。这样它们在累加时操作的不同的 Cell 变量，因此减少了 CAS 重试失败，从而提高性能。

3、Unsafe

$\quad$ Unsafe对象提供了非常底层的，操作内存、线程的方法，Unsafe 对象不能直接调用，只能通过反射获得。

public class TestUnsafe {

    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
        // 通过反射获取unsafe对象
        Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
        theUnsafe.setAccessible(true);
        Unsafe unsafe = (Unsafe) theUnsafe.get(null);

        System.out.println(unsafe);  // sun.misc.Unsafe@85ede7b

        // 利用unsafe对象操作Teacher对象中的属性就行线程安全的操作
        // 1. 获取域的偏移地址
        long idOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("id"));
        long nameOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("name"));

        Teacher t = new Teacher();
        // 2. 执行 cas 操作
        unsafe.compareAndSwapInt(t, idOffset, 0, 1);
        unsafe.compareAndSwapObject(t, nameOffset, null, "cy");

        // 3. 验证
        System.out.println(t);  // Teacher(id=1, name=cy)
    }
}
@Data
class Teacher {
    volatile int id;
    volatile String name;
}

4、不可变类的使用和设计

$\quad$ 可变类都不是线程安全的，不可变类线程安全。比如日期类SimpleDateFormat可变类线程不安全，DateTimeFormatter是不可变类线程安全。
$\quad$ 如何设计不可变类？参考String的设计：

public final class String
    implements java.io.Serializable, Comparable<String>, CharSequence,
               Constable, ConstantDesc {
    @Stable
    private final byte[] value;
    private final byte coder;
    public String(char value[]) {
        this.value = Arrays.copyOf(value, value.length);
    }
    public String substring(int beginIndex, int endIndex) {
        int length = length();
        checkBoundsBeginEnd(beginIndex, endIndex, length);
        if (beginIndex == 0 && endIndex == length) {
            return this;
        }
        int subLen = endIndex - beginIndex;
        return isLatin1() ? StringLatin1.newString(value, beginIndex, subLen)
                          : StringUTF16.newString(value, beginIndex, subLen);
    }
}

类中所有属性都使用final修饰，保证该属性是只读的，不可修改
用final修饰class，保证了该类不能被继承，保证类中方法不能被覆盖，防止子类无意间破坏不可变性
保护性拷贝，在使用char数组初始化String时先拷贝该数组再赋值，直接赋值的话可能会有其他线程在赋值过程中对该数组进行修改，造成线程不安全
subString会创建一个新的字符串返回，不会改变原字符串，与上个例子一样，通过拷贝来避免共享的策略称为保护性共享拷贝

享元模式

$\quad$ 当需要重用数量有限的同一类对象，例如字符串对象，可能很多字符串对象值相同，那么可以共用。享元模式对相同值的对象共享，尽可能节省内存。在包装类中大量使用享元模式：

public static Long valueOf(long l) {
    final int offset = 128;
    if (l >= -128 && l <= 127) { // will cache
        return LongCache.cache[(int)l + offset];
    }
    return new Long(l);
}