高并发编程/消息传递机制避免锁提高并发效率，不懂的赶紧进来（设计篇）

在这里插入图片描述

在现代软件开发中，随着多核处理器的普及和分布式系统的扩展，传统的基于共享内存的并发模型正面临越来越多的挑战。消息传递机制作为一种替代方案，以其独特的异步通信和无共享状态的特性，为构建高效、可扩展和健壮的系统提供了新的思路。它通过将数据操作封装在消息中，允许系统组件以松耦合的方式进行交互，从而减少了锁的需求和竞态条件的风险。本文将深入探讨消息传递机制的原理、优势以及如何在实际应用中实现这一模式，帮助读者理解其在解决并发问题中的重要作用。

肖哥弹架构 跟大家“弹弹” 高并发锁， 关注公号回复 ‘mvcc’ 获得手写数据库事务代码

欢迎点赞，关注，评论。

关注公号Solomon肖哥弹架构获取更多精彩内容

历史热点文章

1、并发问题

1.1 问题描述

在并发环境中，两个线程同时对计数器进行操作，线程1减少2，线程2减少9。由于缺乏同步，两个线程都认为计数器值大于需要减少的值，最终导致计数器变为-1，这违反了业务规则，因为库存不能为负数，表示过度分配。

1.2 解决方案

使用原子操作锁定检查和递减步骤，确保操作的原子性。
- 因为传统并发模式中，共享内存是倾向于强一致性弱隔离性的，例如悲观锁同步的方式就是使用强一致性的方式控制并发，
采用消息传递机制代替共享内存，减少锁的使用。

使用共享数据的并发编程面临的最大问题是数据条件竞争data race，消息传递机制最大的优势在于不会产生数据竞争状态。而实现消息传递有两种常见类型：基于channel的消息传递、基于Actor的消息传递。

1.3 为什么消息传递机制能减少锁

消息传递机制能够减少或消除对锁的需求，主要是因为它改变了并发编程的范式，从直接操作共享状态转变为通过消息传递来协调操作。以下是消息传递机制如何实现这一点的几个关键点：

分解任务：
- 在消息传递模型中，复杂的任务被分解成一系列更小的、可以独立处理的任务单元（消息）。这些任务单元被发送到消息队列中，而不是直接操作共享状态。
无共享状态：
- 每个线程或进程处理自己的任务单元，而不直接访问或修改共享状态。这样，就避免了多个线程同时修改同一共享变量的情况，从而减少了锁的需求。
消费者处理：
- 消费者线程从消息队列中取出任务单元进行处理。由于每个任务单元是独立的，消费者之间不需要同步，因为它们不会同时处理同一个任务单元。
线程安全：
- 消息队列本身是线程安全的，它保证了消息的顺序性和原子性，确保了消息的正确传递和处理。
并发性：
- 由于任务单元是独立的，多个消费者可以并发地从消息队列中取出任务单元进行处理，提高了系统的并发性和吞吐量。
解耦合：
- 消息传递机制使得生产者和消费者之间的耦合度降低，它们不需要知道对方的具体实现，只需要知道如何发送和接收消息。
容错性：
- 如果某个消费者处理任务单元失败，这不会影响其他消费者处理其他任务单元。这种机制提高了系统的容错性。

1.4 消息传递机制的类型

基于Channel的消息传递：在Go语言中广泛使用，通过channel实现goroutine之间的通信。
基于Actor的消息传递：在Akka框架中实现，每个Actor是一个并发执行的实体，通过消息传递进行通信。

1.5 消息传递机制避免锁模型图

在这里插入图片描述

说明：

生产者(Producer) ：在业务逻辑中，当需要减少库存时，生产者将减少库存的请求封装成一条消息，并发送到消息队列中，而不是直接操作共享库存状态。
消息队列(Message Queue) ：消息队列是生产者和消费者之间的中介，它负责存储和传递消息。在这个例子中，消息队列确保了消息的顺序性和独立性，使得每个减少库存的请求都是独立的。
消费者(Consumer) ：消费者从消息队列中取出消息，并根据消息内容执行相应的操作（在这个例子中是减少库存）。由于每个消息都是独立的，消费者不需要与生产者或其他消费者同步，因此避免了锁的使用。

优势：

无共享状态：库存状态不再被多个线程共享，每个减少库存的操作都是通过消息传递来协调的。
线程安全：由于消费者处理的是消息队列中的消息，而不是直接操作共享状态，因此不需要使用锁来保证线程安全。
并发性：多个生产者可以并发地发送消息，多个消费者也可以并发地从消息队列中取出和处理消息，提高了系统的并发处理能力。

1.6 消息传递机制避免锁设计案例

业务：库存管理

假设我们有一个在线商店，需要管理商品的库存。在高并发环境下，多个客户可能同时尝试购买同一件商品，这就要求我们确保库存的减少是线程安全的，以避免库存变为负数。

传统解决方案（使用锁）

在传统的解决方案中，我们可能会使用一个共享的库存计数器，并在减少库存的方法上加上同步锁：

public class Inventory {
    private int stock = 100;

    public synchronized void reduceStock(int amount) {
        if (stock >= amount) {
            stock -= amount;
        } else {
            throw new IllegalArgumentException("库存不足");
        }
    }

    public synchronized int getStock() {
        return stock;
    }
}

在这个例子中，reduceStock 和 getStock 方法都被声明为 synchronized，确保了在同一时间只有一个线程可以修改或读取库存。

使用消息传递机制的解决方案

现在，让我们使用消息传递机制来重构这个库存管理的业务逻辑，避免使用锁：

import java.util.concurrent.ConcurrentLinkedQueue;

public class InventoryManager {
    private final ConcurrentLinkedQueue<InventoryCommand> commandQueue = new ConcurrentLinkedQueue<>();

    public void processCommands() {
        while (!Thread.currentThread().isInterrupted()) {
            InventoryCommand command = commandQueue.poll();
            if (command != null) {
                command.execute();
            }
        }
    }

    public void reduceStock(int amount) {
        commandQueue.offer(new InventoryCommand(amount));
    }

    private static class InventoryCommand {
        private final int amount;
        private int stock = 100; // 每个命令有自己的库存副本

        public InventoryCommand(int amount) {
            this.amount = amount;
        }

        public void execute() {
            if (stock >= amount) {
                stock -= amount;
                System.out.println("库存减少 " + amount + "，当前库存 " + stock);
            } else {
                System.out.println("库存不足，无法减少 " + amount);
            }
        }
    }
}

public class Main {
    public static void main(String[] args) {
        InventoryManager manager = new InventoryManager();
        Thread commandProcessor = new Thread(manager::processCommands);
        commandProcessor.start();

        // 模拟多个线程减少库存
        for (int i = 0; i < 5; i++) {
            int finalI = i;
            new Thread(() -> manager.reduceStock(20)).start();
        }

        // 等待命令处理
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        commandProcessor.interrupt();
    }
}

解释

在这个改进的例子中：

InventoryCommand 是一个包含库存减少逻辑的类，每个命令都有自己的库存副本。这意味着每个命令处理自己的库存状态，而不是共享一个全局的库存状态。
reduceStock 方法将减少库存的操作封装为一个 InventoryCommand 对象，并将其添加到命令队列中。
processCommands 方法从队列中取出命令并执行，由于每个命令处理自己的库存副本，因此不需要使用锁。
这里private int stock = 100;定义在InventoryCommand类中，使得每个InventoryCommand对象都有自己的库存副本，这样做的主要目的是为了避免锁的使用，并实现以下几个关键点：
1. 无共享状态：
  - 每个InventoryCommand对象管理自己的库存状态，不依赖于全局共享的库存状态。这意味着不同的消息（命令）之间不会直接竞争或冲突，因为它们各自操作自己的数据副本。
2. 线程安全：
  - 由于每个命令操作的是自己的库存副本，不存在多个线程同时修改同一共享变量的情况，从而避免了并发修改导致的数据不一致问题，也就不需要使用锁来保证线程安全。
3. 简化并发控制：
  - 在传统的并发编程中，通常需要使用锁（如synchronized块或ReentrantLock）来保护对共享资源的访问。通过为每个任务提供独立的数据副本，可以避免这些复杂的并发控制机制，简化编程模型。
4. 提高性能和可扩展性：
  - 避免使用锁可以减少线程间的协调开销，提高系统的吞吐量和响应性。在多核处理器上，无锁的设计可以更好地利用硬件资源，提高并行处理能力。
5. 容错性：
  - 在消息传递模型中，每个消息（命令）的处理是独立的，一个命令的失败不会影响到其他命令的执行，从而提高了系统的容错性。

替代方案：使用不可变对象

另一种避免锁的方法是使用不可变对象。不可变对象一旦创建，其状态就不能被改变，因此天生是线程安全的，不需要使用锁。例如，我们可以定义一个不可变的库存命令对象：

public final class InventoryCommand {
    private final int amount;
    private final int newStock;

    public InventoryCommand(int amount, int currentStock) {
        this.amount = amount;
        this.newStock = currentStock - amount;
    }

    public int getNewStock() {
        return newStock;
    }

    public int getAmount() {
        return amount;
    }
}

在这个版本中，InventoryCommand对象在创建时就计算了新的库存值，并且这个值是不可变的。处理命令时，我们只需读取命令的属性，而不需要修改它：

public void processCommands() {
    while (!Thread.currentThread().isInterrupted()) {
        InventoryCommand command = commandQueue.poll();
        if (command != null) {
            int newStock = command.getNewStock();
            System.out.println("库存减少 " + command.getAmount() + "，当前库存 " + newStock);
        }
    }
}

这种方法进一步简化了设计，因为命令对象本身不包含任何可变状态，从而完全避免了锁的需求。

1.7. 结论

消息传递机制通过改变并发编程的范式，从直接操作共享状态转变为通过消息传递来协调操作，从而减少了锁的使用，提高了系统的并发性和容错性。这种机制特别适用于需要高吞吐量和高可靠性的分布式系统。

标签：库存,int,编程,并发,amount,线程,消息传递
From： https://blog.csdn.net/alises1314/article/details/143757767