使用JDK提供的常用工具在多线程编写线程安全和数据同步的程序

原创

codetrend

发布于 2024-03-27 09:05:40

1250

发布于 2024-03-27 09:05:40

文章被收录于专栏：高并发编程

题图来自APOD

你好，这里是codetrend专栏“高并发编程基础”。

引言

在并发执行任务时，由于资源共享的存在，线程安全成为一个需要考虑的问题。与串行化程序相比，并发执行可以更好地利用CPU计算能力，提高系统的吞吐量。

例如，当B客户提交一个业务请求时，不需要等到A客户处理结束才能开始，这样可以提升用户体验。

然而，并发编程也带来了新的挑战。无论是互联网系统还是企业级系统，在追求高性能的同时，稳定性也是至关重要的。开发人员需要掌握高效编程的技巧，以确保程序在安全的前提下能够高效地共享数据。

共享资源指多个线程同时对同一份资源进行读写操作，这就需要保证多个线程访问到的数据是一致的，即数据同步或资源同步。为了实现安全且高效的共享数据，以下是一些常用的方法和技术：

使用锁（Lock）：通过使用锁机制，只有获得锁的线程才能访问共享资源，其他线程需要等待锁的释放。常见的锁包括synchronized关键字、ReentrantLock等。锁机制可以保证共享资源在同一时间只被一个线程访问，从而避免数据竞争和不一致的问题。
使用同步块（Synchronized Block）：通过在代码块前加上synchronized关键字，确保同一时间只有一个线程可以执行该代码块。这样可以限制对共享资源的访问，保证数据的一致性。
使用原子操作类（Atomic Classes）：Java提供了一系列原子操作类，如AtomicInteger、AtomicLong等，它们可以保证针对共享资源的操作是原子性的，不会被其他线程中断，从而避免了数据不一致的问题。
使用并发集合（Concurrent Collections）：Java提供了一些并发安全的集合类，如ConcurrentHashMap、ConcurrentLinkedQueue等，它们在多线程环境下可以安全地进行读写操作，避免了手动处理同步和锁的麻烦。
使用线程安全的设计模式：在程序设计阶段，可以采用一些线程安全的设计模式，如不可变对象、线程本地存储(Thread-local Storage)等，来避免共享资源的竞争和冲突。

数据不一致的问题

package engineer.concurrent.battle.abasic;

/**
 * 叫号机排队模拟，通过多线程并发
 */
public class TicketWindow extends Thread {
    private final String name;
    private final static int MAX = 100;
    private static int ticket = 1;

    public TicketWindow(String name) {
        this.name = name;
    }
    public void run() {
        while (ticket<= MAX) {
            System.out.println(name + "柜台正在排队，排队号码为：" + ticket);
            ticket++;
        }
    }
    public static void main(String[] args) {
        new TicketWindow("一号窗口").start();
        new TicketWindow("二号窗口").start();
        new TicketWindow("三号窗口").start();
        new TicketWindow("四号窗口").start();
    }
}

可能的输出结果如下：

三号窗口柜台正在排队，排队号码为：1
四号窗口柜台正在排队，排队号码为：1
四号窗口柜台正在排队，排队号码为：3
三号窗口柜台正在排队，排队号码为：2
四号窗口柜台正在排队，排队号码为：7
...
四号窗口柜台正在排队，排队号码为：101
四号窗口柜台正在排队，排队号码为：102

其中 ticket 就是共享资源，多个TicketWindow运行多线程竞争共享资源。可能出现的问题如下。

ticket 被重复使用，也就是一个号被多个窗口叫到。
ticket 超过最大限制，也就是实际没得这个号但是却叫号了。
ticket 没有被使用，也就是一张号没有被叫到。

下面的实例代码是叫号机排队模拟，通过多线程并发，使用synchronized解决资源共享问题

package engineer.concurrent.battle.esafe;

import java.util.concurrent.TimeUnit;

/**
 * 叫号机排队模拟，通过多线程并发，使用synchronized解决资源共享问题
 */
public class TicketWindowSynchronized implements Runnable {
    private final static int MAX = 100;
    private static Integer ticket = 1;
    private static final Object lockObj = new Object();

    public void run() {
        while (ticket <= MAX) {
            synchronized (lockObj) {
                if (ticket <= MAX) { // 额外的判断
                    System.out.println(Thread.currentThread() + "柜台正在排队，排队号码为：" + ticket);
                    ticket++;
                }
            }
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void main(String[] args) {
        TicketWindowSynchronized ticketTask = new TicketWindowSynchronized();
        new Thread(ticketTask, "一号窗口").start();
        new Thread(ticketTask, "二号窗口").start();
        new Thread(ticketTask, "三号窗口").start();
        new Thread(ticketTask, "四号窗口").start();
    }
}

使用同步块（Synchronized Block）

在Java中，同步块（Synchronized Block）是一种用于实现线程同步的机制。它用于标记一段代码，确保在同一时间只有一个线程可以执行这段代码，以避免数据竞争和并发问题。synchronized 字段可以用于对象方法、代码块中。

同步实例方法：

public synchronized void synchronizedMethod() {
    // 执行需要同步的代码
}

在实例方法上使用synchronized关键字，将整个方法体标记为同步块。当一个线程进入同步方法时，它将获取该实例对象的锁，并且其他线程将被阻塞等待锁的释放。

同步静态方法：

public static synchronized void synchronizedStaticMethod() {
    // 执行需要同步的代码
}

在静态方法上使用synchronized关键字，将整个静态方法标记为同步块。与同步实例方法类似，当一个线程进入同步静态方法时，它将获取类对象的锁，并且其他线程将被阻塞等待锁的释放。

同步块：

synchronized (lockObj) {
    // 执行需要同步的代码
}

使用synchronized关键字结合一个对象来创建同步块。当一个线程进入同步块时，它将获取该对象的锁，并且其他线程将被阻塞等待锁的释放。在同步块内，只有一个线程可以执行被同步的代码。

同步块中的条件等待和唤醒：

synchronized (lockObj) {
    while (!conditionMet) {
        try {
            lockObj.wait(); // 条件不满足时，线程进入等待状态并释放锁
        } catch (InterruptedException e) {
            // 处理中断异常
        }
    }
    // 执行需要同步的代码
}

synchronized (lockObj) {
    conditionMet = true; // 修改条件
    lockObj.notify(); // 唤醒一个等待的线程
    lockObj.notifyAll(); // 唤醒所有等待的线程
}

在同步块中，使用对象的wait()方法让线程进入等待状态并释放锁。当某个条件满足时，可以使用notify()或notifyAll()方法唤醒等待的线程。注意，在使用条件等待和唤醒时，需要确保线程在同一对象上等待和唤醒。

同步块提供了一种简单的方式来实现线程同步，通过获取对象的锁来保证同一时间只有一个线程可以执行同步块内的代码。这对于控制并发访问共享资源非常有用。但是需要注意，如果多个线程竞争相同的锁，可能会导致性能问题和死锁情况的发生。因此，在使用同步块时，需要仔细考虑锁的粒度和设计。

测试代码如下：

package engineer.concurrent.battle.esafe;
public class SynchronizedCounter {
    private int c = 0;

    public synchronized void increment() {
        System.out.println(Thread.currentThread());;
        c++;
    }

    public synchronized void decrement() {
        c--;
    }

    public synchronized int value() {
        return c;
    }
}

package engineer.concurrent.battle.esafe;

import java.util.concurrent.CountDownLatch;

public class SynchronizedCounterTest {
    public static void main(String[] args) throws InterruptedException {
        SynchronizedCounter counter = new SynchronizedCounter();
        CountDownLatch countDownLatch = new CountDownLatch(1000);
        for (int i = 0; i < 1000; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    counter.increment();
                }
                countDownLatch.countDown();
            },"线程编号"+i).start();
        }
        countDownLatch.await();
        System.out.println(counter.value());
    }
}

输出结果如下：

使用锁（Lock）

在Java中，锁（Lock）是一种用于实现线程同步的机制。它可以确保在同一时间只有一个线程可以访问共享资源，以避免数据竞争和并发问题。与传统的synchronized关键字相比，Lock提供了更大的灵活性和功能。使用锁（Lock）机制可以更细粒度地控制线程同步，并且提供了更多高级功能，例如可中断的锁获取、定时锁获取和条件变量等待。这使得锁成为Java中多线程编程的重要组件之一。

创建Lock对象：

Lock lock = new ReentrantLock();

获取锁：

lock.lock(); // 如果锁可用，获取锁；否则等待锁的释放

或者带有超时设置的获取锁：

boolean acquired = lock.tryLock(5, TimeUnit.SECONDS); // 尝试在指定时间内获取锁，返回是否成功获取锁
if (acquired) {
    try {
        // 执行需要同步的代码
    } finally {
        lock.unlock(); // 释放锁
    }
} else {
    // 获取锁失败的处理逻辑
}

释放锁：

lock.unlock(); // 释放锁

使用锁进行同步：

lock.lock();
try {
    // 执行需要同步的代码
} finally {
    lock.unlock();
}

使用锁的Condition进行条件等待和唤醒：

Condition condition = lock.newCondition();

// 等待条件满足
lock.lock();
try {
    while (!conditionMet) {
        condition.await(); // 等待条件满足并释放锁
    }
    // 执行需要同步的代码
} catch (InterruptedException e) {
    // 处理中断异常
} finally {
    lock.unlock();
}

// 唤醒等待的线程
lock.lock();
try {
    condition.signal(); // 唤醒一个等待的线程
    condition.signalAll(); // 唤醒所有等待的线程
} finally {
    lock.unlock();
}

测试代码如下：

package engineer.concurrent.battle.esafe;

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

public class SynchronizedCounter3 {
    private int c = 0;
    Lock lock = new ReentrantLock();

    public void increment() {
        lock.lock();
        System.out.println(Thread.currentThread());
        c++;
        lock.unlock();
    }

    public void decrement() {
        lock.lock();
        c--;
        lock.unlock();
    }

    public int value() {
        return c;
    }
}

package engineer.concurrent.battle.esafe;

import java.util.concurrent.CountDownLatch;

public class SynchronizedCounter3Test {
    public static void main(String[] args) throws InterruptedException {
        SynchronizedCounter3 counter = new SynchronizedCounter3();
        CountDownLatch countDownLatch = new CountDownLatch(10000);
        for (int i = 0; i < 10000; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    counter.increment();
                }
                countDownLatch.countDown();
            },"线程编号"+i).start();
        }
        countDownLatch.await();
        System.out.println(counter.value());
    }
}

输出结果如下，在并发情况下输出一致：

使用原子操作类（Atomic Classes）

在Java中，原子操作类（Atomic Classes）是一组线程安全的工具类，用于进行原子性操作。它们提供了一些原子操作，可以确保在多线程环境下对共享变量的操作是原子的，不会出现数据竞争和并发问题。原子操作类提供了一些常见的原子操作方法，可以确保对共享变量的操作是原子的。它们适用于高并发场景，并且性能较好。使用原子操作类可以避免使用锁带来的开销，并且能够简化线程同步的代码逻辑。

需要注意的是，虽然原子操作类可以保证单个操作的原子性，但不能保证多个操作的原子性。如果需要进行复合操作，例如读取-修改-写入操作，仍然需要使用锁或其他同步机制来保证原子性。另外，原子操作类在某些情况下可能会存在ABA问题，需要根据具体场景选择合适的解决方案。

AtomicBoolean：

AtomicBoolean atomicBoolean = new AtomicBoolean();

boolean currentValue = atomicBoolean.get(); // 获取当前值

atomicBoolean.set(true); // 设置新值

boolean oldValue = atomicBoolean.getAndSet(false); // 先获取当前值，再设置新值，并返回旧值

AtomicInteger：

AtomicInteger atomicInteger = new AtomicInteger();

int currentValue = atomicInteger.get(); // 获取当前值

atomicInteger.set(10); // 设置新值

int oldValue = atomicInteger.getAndSet(5); // 先获取当前值，再设置新值，并返回旧值

int newValue = atomicInteger.incrementAndGet(); // 原子地增加1，并返回新值

int updatedValue = atomicInteger.updateAndGet(x -> x * 2); // 使用lambda表达式更新值，并返回更新后的值

AtomicLong：

AtomicLong atomicLong = new AtomicLong();

long currentValue = atomicLong.get(); // 获取当前值

atomicLong.set(100L); // 设置新值

long oldValue = atomicLong.getAndSet(50L); // 先获取当前值，再设置新值，并返回旧值

long newValue = atomicLong.incrementAndGet(); // 原子地增加1，并返回新值

long updatedValue = atomicLong.updateAndGet(x -> x * 2); // 使用lambda表达式更新值，并返回更新后的值

AtomicReference：

AtomicReference<String> atomicReference = new AtomicReference<>();

String currentValue = atomicReference.get(); // 获取当前值

atomicReference.set("Hello"); // 设置新值

String oldValue = atomicReference.getAndSet("World"); // 先获取当前值，再设置新值，并返回旧值

boolean updated = atomicReference.compareAndSet("World", "Java"); // 原子地比较和设置值，返回是否成功更新

测试代码如下：

package engineer.concurrent.battle.esafe;

import java.util.concurrent.atomic.AtomicInteger;

public class SynchronizedCounter2 {
    private AtomicInteger c = new AtomicInteger(0);

    public synchronized void increment() {
        System.out.println(Thread.currentThread());
        c.incrementAndGet();
    }

    public void decrement() {
        c.decrementAndGet();
    }

    public int value() {
        return c.get();
    }
}

package engineer.concurrent.battle.esafe;

import java.util.concurrent.CountDownLatch;

public class SynchronizedCounter2Test {
    public static void main(String[] args) throws InterruptedException {
        SynchronizedCounter2 counter = new SynchronizedCounter2();
        CountDownLatch countDownLatch = new CountDownLatch(1000);
        for (int i = 0; i < 1000; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    counter.increment();
                }
                countDownLatch.countDown();
            },"线程编号"+i).start();
        }
        countDownLatch.await();
        System.out.println(counter.value());
    }
}

输出结果如下，在并发情况下输出一致：

使用并发集合（Concurrent Collections）

在Java中，有一组并发集合（Concurrent Collections）可以用于在多线程环境下安全地操作共享数据。这些集合类提供了线程安全的操作，并且能够处理高并发的情况，常用于多线程编程和并发控制。并发集合提供了一些常见的数据结构和操作方法，能够在多线程环境下安全地进行读写操作。它们采用了特定的并发控制策略，以提供高效的线程安全性能。需要根据具体的场景选择合适的并发集合类，以满足线程安全和并发控制的需求。

需要注意的是，并发集合并不适用于所有情况。在某些场景下，例如需要保持原子性操作或依赖复合操作的情况下，可能需要使用其他的同步机制来确保线程安全性。此外，虽然并发集合可以提供更好的性能和扩展性，但在某些情况下可能会占用更多的内存，需要根据具体情况进行权衡和选择。

ConcurrentHashMap：

ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

map.put("key1", 1); // 插入键值对

int value = map.get("key1"); // 获取指定键的值

boolean containsKey = map.containsKey("key2"); // 检查是否包含指定的键

Integer oldValue = map.putIfAbsent("key1", 2); // 当键不存在时才插入新值

map.remove("key1"); // 移除指定键的键值对

ConcurrentLinkedQueue：

ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();

queue.add("item1"); // 添加元素到队列

String head = queue.peek(); // 获取队列头部元素

String removedItem = queue.poll(); // 移除并返回队列头部元素

CopyOnWriteArrayList：

CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();

list.add("item1"); // 添加元素到列表

String item = list.get(0); // 获取指定索引处的元素

list.set(0, "newItem"); // 替换指定索引处的元素

boolean removed = list.remove("item1"); // 移除指定元素

ConcurrentSkipListMap：

ConcurrentSkipListMap<Integer, String> map = new ConcurrentSkipListMap<>();

map.put(1, "value1"); // 插入键值对

Integer key = map.firstKey(); // 获取第一个键

String value = map.get(key); // 根据键获取值

map.remove(key); // 移除指定键的键值对

测试代码如下：

package engineer.concurrent.battle.esafe;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class SynchronizedCounter5 {
    private ConcurrentMap<String, Integer> counter = new ConcurrentHashMap<>();
    private final String key = "threadName";

    public void increment() {
        counter.compute(key, (key, value) -> (value == null) ? 1 : value + 1);
    }
    public void decrement() {
        counter.compute(key, (key, value) -> (value != null && value > 0) ? value - 1 : 0);
    }
    public int value() {
        return counter.values().stream().mapToInt(Integer::intValue).sum();
    }
}

package engineer.concurrent.battle.esafe;

import java.util.concurrent.CountDownLatch;

public class SynchronizedCounterTest5 {
    public static void main(String[] args) throws InterruptedException {
        SynchronizedCounter5 counter = new SynchronizedCounter5();
        CountDownLatch countDownLatch = new CountDownLatch(1000);
        for (int i = 0; i < 1000; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    counter.increment();
                }
                countDownLatch.countDown();
            },"线程编号"+i).start();
        }
        countDownLatch.await();
        System.out.println(counter.value());
    }
}

输出结果如下，在并发情况下输出一致：

死锁原因和分析

死锁的产生

因为线程中锁的加入和线程同步到需求存在，资源的竞争问题解决了，但问题出现在解决办法（也就是锁）的不合理使用会导致死锁的出现。死锁是多线程编程中常见的问题，指两个或多个线程因为互相持有对方需要的锁而陷入了无限等待的状态。Java中的死锁通常发生在如下情况下：

竞争有限资源：多个线程同时竞争一些有限的资源，例如数据库连接、文件句柄等。
锁嵌套：一个线程持有一个锁，尝试获取另一个锁，而另一个线程持有第二个锁并尝试获取第一个锁。

下面是一个造成死锁的示例代码：

/**
 * 在示例代码中，两个线程分别持有 lock1 和 lock2，并尝试获取对方持有的锁。如果这两个线程同时运行，就会发生死锁，因为它们互相持有了对方需要的锁。
 */
public class DeadlockExample {
    static Object lock1 = new Object();
    static Object lock2 = new Object();

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            synchronized (lock1) { // 获取 lock1
                System.out.println("Thread 1: Holding lock 1...");

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {}

                synchronized (lock2) { // 尝试获取 lock2
                    System.out.println("Thread 1: Holding lock 1 & 2...");
                }
            }
        },"线程001");

        Thread t2 = new Thread(() -> {
            synchronized (lock2) { // 获取 lock2
                System.out.println("Thread 2: Holding lock 2...");

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {}

                synchronized (lock1) { // 尝试获取 lock1
                    System.out.println("Thread 2: Holding lock 1 & 2...");
                }
            }
        },"线程002");

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

死锁的分析

通过jdk提供的开箱即用工具可以快速定位问题。以下是DeadlockExample产生死锁的定位过程。

使用 jps 查看当前进程的pid。

jps
20580 Launcher
44712 RemoteMavenServer36
12428 Jps
14556
25052 DeadlockExample

使用 jstack pid命令分析堆栈信息。输出结果如下。

Found one Java-level deadlock:
=============================
"线程001":
  waiting to lock monitor 0x0000019d2d1eb820 (object 0x0000000713cfa108, a java.lang.Object),
  which is held by "线程002"

"线程002":
  waiting to lock monitor 0x0000019d2d1eb740 (object 0x0000000713cfa0f8, a java.lang.Object),
  which is held by "线程001"

Java stack information for the threads listed above:
===================================================
"线程001":
        at engineer.concurrent.battle.esafe.DeadlockExample.lambda$main$0(DeadlockExample.java:20)
        - waiting to lock <0x0000000713cfa108> (a java.lang.Object)
        - locked <0x0000000713cfa0f8> (a java.lang.Object)
        at engineer.concurrent.battle.esafe.DeadlockExample$$Lambda$14/0x0000000800c01200.run(Unknown Source)
        at java.lang.Thread.run(java.base@17.0.7/Thread.java:833)
"线程002":
        at engineer.concurrent.battle.esafe.DeadlockExample.lambda$main$1(DeadlockExample.java:34)
        - waiting to lock <0x0000000713cfa0f8> (a java.lang.Object)
        - locked <0x0000000713cfa108> (a java.lang.Object)
        at engineer.concurrent.battle.esafe.DeadlockExample$$Lambda$15/0x0000000800c01418.run(Unknown Source)
        at java.lang.Thread.run(java.base@17.0.7/Thread.java:833)

Found 1 deadlock.

上面的jstack信息清晰的给出了死锁的代码位置和线程名称。通过这两个信息可以定位到代码块进行对应问题的修复。

死锁的避免

要避免死锁，可以采取以下策略：

避免锁嵌套：尽量减少锁嵌套的层数，以避免死锁的发生。
按固定顺序获取锁：多个线程按照固定的顺序获取锁，以避免交叉竞争造成的死锁。
使用 tryLock() 方法：tryLock() 方法可以尝试获取锁一段时间，如果失败则放弃获取锁，避免一直等待造成的死锁。
使用 LockInterruptibly() 方法：LockInterruptibly() 方法可以在等待锁的过程中响应中断信号，避免无限等待造成的死锁。
合理设计资源分配：合理地划分和分配资源，以避免资源争用和死锁的产生。

修改后的代码如下：

package engineer.concurrent.battle.esafe;

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

/**
 * 使用 tryLock() 方法：tryLock() 方法可以尝试获取锁一段时间，如果失败则放弃获取锁，避免一直等待造成的死锁。
 */
public class DeadlockExampleFix {
    static Lock lock1 = new ReentrantLock();
    static Lock lock2 = new ReentrantLock();

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            if(lock1.tryLock()) { // 获取 lock1
                System.out.println("Thread 1: Holding lock 1...");

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {}

                if(lock2.tryLock()) { // 尝试获取 lock2
                    System.out.println("Thread 1: Holding lock 1 & 2...");
                }
            }
        },"线程001");

        Thread t2 = new Thread(() -> {
            if(lock2.tryLock()) { // 获取 lock2
                System.out.println("Thread 2: Holding lock 2...");

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {}

                if(lock1.tryLock()) { // 尝试获取 lock1
                    System.out.println("Thread 2: Holding lock 1 & 2...");
                }
            }
        },"线程002");

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

输出结果如下：

Thread 2: Holding lock 2...
Thread 1: Holding lock 1...

参考

《Java高并发编程详解：多线程与架构设计》

关于作者

来自一线全栈程序员nine的探索与实践，持续迭代中。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

多线程

高并发

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

多线程

高并发

登录后参与评论

0 条评论

热度