关于程序员:分布式事务的21种武器-1

在分布式系统中，事务的解决散布在不同组件、服务中，因而分布式事务的 ACID 保障面临着一些非凡难点。本系列文章介绍了 21 种分布式事务设计模式，并剖析其实现原理和优缺点，在面对具体分布式事务问题时，能够抉择适合的模式进行解决。原文: Exploring Solutions for Distributed Transactions (1)”)

因为良好的可伸缩性和容错性，分布式系统当初变得越来越广泛，但在分布式系统中保护数据一致性可能十分艰难，尤其是在在须要解决跨多个节点的事务时。接下来咱们就来探讨那些能够确保分布式系统数据一致性的技术、模式和算法。

分布式事务波及多个节点，它们一起工作以执行单个事务。确保所有节点对事务后果达成统一，同时保持数据一致性是一项具备挑战性的工作。传统事务的 ACID(原子性、一致性、隔离性、持久性)属性在分布式环境中变得更加难以实现。

本文旨在全面介绍可用于保护分布式系统数据一致性的不同办法，探讨每种办法的优缺点，以及在什么场景下实用。

在不同业务场景下，能够有不同的解决方案，常见办法有:

阻塞重试(Blocking Retry)
二阶段和三阶段提交(Two-Phase Commit (2PC) and Three-Phase Commit (3PC))
基于后盾队列的异步解决(Using Queues to Process Asynchronously in the Background)
TCC 弥补(TCC Compensation Matters)
本地音讯表(异步保障)/ 发件箱模式(Local Message Table (Asynchronously Ensured)/Outbox Pattern)
MQ 事务(MQ Transaction)
Saga 模式(Saga Pattern)
事件驱动(Event Sourcing)
命令查问职责拆散(Command Query Responsibility Segregation, CQRS)
原子提交(Atomic Commitment)
并行提交(Parallel Commits)
事务复制(Transactional Replication)
一致性算法(Consensus Algorithms)
工夫戳排序(Timestamp Ordering)
乐观并发管制(Optimistic Concurrency Control)
拜占庭容错(Byzantine Fault Tolerance, BFT)
分布式锁(Distributed Locking)
分片(Sharding)
多版本并发管制(Multi-Version Concurrency Control, MVCC)
分布式快照(Distributed Snapshots)
主从复制(Leader-Follower Replication)

本文将介绍阻塞重试、2PC/3PC、后盾队列三种模式。

1. 阻塞重试(Blocking Retry)

解决与近程服务或资源交互时可能产生的谬误或失败。
当产生谬误时，能够主动重试，通常在重试之前会期待一段时间，重试直到操作胜利或达到最大重试次数为止。
业务执行代码同步期待操作实现(胜利或失败)，而后再尝试重试操作。这意味着执行操作的线程在期待操作实现时被阻塞，在操作实现或失败之前，不能在该线程上执行其余工作。
在 Python 中应用 retrying 库实现阻塞重试。

import retrying

@retrying.retry(wait_fixed=1000, stop_max_delay=10000)
def my_function():
    # code to retry goes here
    # if this function throws an exception, it will be retried according to the parameters specified above

示例代码:

应用 retrying 库实现阻塞重试行为。
用 @retrying.retry 装璜想要重试的函数，并传入参数指定重试行为逻辑。
用 wait_fixed 参数指定重试之间期待 1 秒，用 stop_max_delay 参数指定在 10 秒后进行重试。
如果 my_function 抛出异样，将会触发重试，直到胜利或 10 秒超时为止。

长处

通过应用程序或零碎主动重试失败的操作，直到操作胜利，从而进步应用程序或零碎的可靠性，避免由刹时问题引起的谬误和故障。
缩小停机工夫，确保要害系统对用户的可用性。
主动解决失败操作，缩小系统管理员和反对团队的工作量。

毛病

当重试很频繁或须要很长时间能力实现时，会在零碎中引入额定提早。
当重试机制比拟敏感并且频繁重试时，会耗费额定资源。
阻塞重试机制可能陷入有限循环或导致死锁，导致一直重试失败的操作而无奈胜利。

实用场景

在执行的操作十分要害且必须胜利实现的场景中十分有用。
电子商务 —— 能够用来主动重试失败的交易，如购买或退款，以进步可靠性并缩小支出损失的危险。
企业软件 —— 能够主动重试企业软件中失败的数据库或 API 操作，进步可靠性并缩小 IT 反对团队的工作量。
云应用程序 —— 阻塞重试机制可用于主动重试失败的云 API 申请或服务调用，进步可靠性并升高基于云的应用程序的停机危险。

挑战

确定适当的重试参数，如重试次数、重试距离和回退策略。
解决无奈通过重试机制主动解决的谬误，例如由网络问题、数据损坏或有效输出引起的谬误。
跨多个节点协调重试可能具备挑战性，特地是当节点具备不同的故障模式或重试机制时。

2. 二阶段和三阶段提交(Two-Phase Commit (2PC) and Three-Phase Commit (3PC))

两阶段提交 (two-phase commit, 2PC) 协定是一种分布式算法，用于确保分布式系统中多个数据库或资源之间事务的原子性和一致性。
第一阶段，协调器向事务中所有参与者发送音讯，要求他们为提交做筹备。而后，每个参与者将在本地执行事务，并以 ”yes”(示意筹备提交)或 ”no”(示意不能提交)投票进行响应。
第二阶段，如果所有参与者投票 ”yes”，协调器将向所有参与者发送提交音讯，要求他们提交事务。如果有参与者投票 ”no”，或者协调器在指定工夫内没有收到所有参与者的响应，协调器将向所有参与者发送停止音讯，要求他们回滚事务。
提交操作要么胜利实现，要么终止，所有参与者都会收到相应告诉。协调器必须期待直到收到来自所有参与者的确认，能力发表事务实现。

# Coordinator code
def two_phase_commit(coordinator, participants, data):
    # Phase 1: Prepare phase
    prepare_ok = True
    for participant in participants:
        try:
            participant.prepare(data)
        except:
            prepare_ok = False
            break
    if not prepare_ok:
        # Phase 1b: Abort
        for participant in participants:
            participant.abort()
        return False
    else:
        # Phase 2: Commit phase
        commit_ok = True
        for participant in participants:
            try:
                participant.commit()
            except:
                commit_ok = False
                break
        if not commit_ok:
            # Phase 2b: Rollback
            for participant in participants:
                participant.rollback()
            return False
        else:
            return True

# Participant code
class Participant:
    def prepare(self, data):
        # Perform prepare operations on local data
        # If prepare fails, raise an exception
        pass

    def commit(self):
        # Perform commit operations on local data
        pass

    def rollback(self):
        # Perform rollback operations on local data
        pass

示例代码:

波及 2 个组件(协调器和参与者)
协调器向参与者发送 ” 筹备(prepare)” 申请，以确保参与者曾经筹备好提交事务。如果参与者答复 ”Yes”，协调器将向参与者发送 ” 提交(commit)” 申请。如果参与者再次响应 ”Yes”，协调器将向参与者和协调器发送全局提交音讯。
如果任何参与者响应 ”No” 或在处理过程中呈现谬误，协调器将向参与者和协调器发送停止音讯以回滚事务。

长处

确保事务的原子性和一致性
提供了解决失败以及停止事务的机制
实现多节点的协调，而不须要人工干预

毛病

可能须要很长时间期待所有节点的确认，并可能导致阻塞
协调器节点可能造成单点故障
实现可能比较复杂，须要认真设计以防止死锁和竞态条件

实用场景

事务波及多个账户和零碎的银行、金融零碎
须要跨多个仓库和零碎更新库存的电子商务系统
订单须要跨多个供应商和零碎进行解决和跟踪的供应链管理系统

挑战

确保所有参加节点都接管到 ” 筹备 ” 和 ” 提交 ” 音讯
告诉所有参加节点停止事务
确保协调器节点可用并失常工作

三阶段提交(3PC) 是 2PC 协定的扩大，解决了 2PC 中筹备阶段的阻塞。3PC 引入了第三个阶段，称为 ” 预提交(pre-commit)” 阶段，升高了阻塞的可能性。
CanCommit 阶段 —— 事务协调器向每个参与者发送音讯，询问是否筹备好提交事务。如果参与者筹备好了，向协调器发送 ”Yes” 音讯，如果没有筹备好，发送 ”No” 音讯。如果所有参与者都答复 ”Yes”，协调器将进入预提交阶段。然而，如果一个或多个参与者响应 ”No”，协调器立刻向所有参与者发送停止音讯，回滚事务。
Pre-Commit 阶段 —— 协调器向所有参与者发送预提交音讯，为提交做筹备。如果所有参与者都确认预提交音讯，协调器将切换到 DoCommit 阶段。然而，如果一个或多个参与者没有响应，协调器就认为失败了，并向所有参与者发送停止音讯。
DoCommit 阶段 —— 协调器向所有参与者发送 Commit 音讯，要求提交事务。如果所有参与者都接管到 Commit 音讯并胜利提交事务，将向协调器发送确认音讯。如果任何参与者未能提交事务，将向协调器发送停止音讯。如果协调器从任何参与者接管到停止音讯，将向所有参与者发送停止音讯，回滚事务。

# Coordinator code
def three_phase_commit(coordinator, participants, data):
    # Phase 1: CanCommit phase
    can_commit_ok = True
    for participant in participants:
        try:
            participant.can_commit(data)
        except:
            can_commit_ok = False
            break
    if not can_commit_ok:
        # Phase 1b: Abort
        for participant in participants:
            participant.abort()
        return False
    else:
        # Phase 2: PreCommit phase
        pre_commit_ok = True
        for participant in participants:
            try:
                participant.pre_commit()
            except:
                pre_commit_ok = False
                break
        if not pre_commit_ok:
            # Phase 2b: Abort
            for participant in participants:
                participant.abort()
            return False
        else:
            # Phase 3: DoCommit phase
            do_commit_ok = True
            for participant in participants:
                try:
                    participant.do_commit()
                except:
                    do_commit_ok = False
                    break
            if not do_commit_ok:
                # Phase 3b: Rollback
                for participant in participants:
                    participant.rollback()
                return False
            else:
                return True

# Participant code
class Participant:
    def can_commit(self, data):
        # Determine if participant can commit
        # If participant cannot commit, raise an exception
        pass

    def pre_commit(self):
        # Perform pre-commit operations on local data
        pass

    def do_commit(self):
        # Perform commit operations on local data
        pass

    def rollback(self):
        # Perform rollback operations on local data
        pass

示例代码:

波及 3 个组件(1 个协调器和 2 个参与者)。
协调器向两个参与者发送 ”can-commit” 音讯，确保曾经筹备好提交事务。如果两个参与者都响应 ”Yes”，协调器将向两个参与者发送 ”pre-commit” 音讯，告诉他们事务行将提交。如果两个参与者都响应确认，协调器将向两个参与者发送 ”commit” 音讯以提交事务。
如果任何参与者响应 ”No” 或在处理过程中呈现谬误，协调器将向两个参与者发送停止音讯以回滚事务。

长处

因为决策过程散布在 3 个阶段，因而缩短了阻塞工夫
在协调器失败的状况下，参与者依然能够做出决定并提交或终止事务
容许多个参与者单独决策，防止了由单个协调器引起的瓶颈

毛病

很难实现和保护
引入额定的音讯替换，可能会减少事务的提早
不是分布式事务的残缺解决方案

实用场景

须要高可用性、容错、波及多方的事务和高可伸缩性的场景
须要高可用性、容错和一致性的金融事务
波及多方的电子商务交易
波及敏感数据的医疗保健事务

挑战

比 2PC 更简单
会导致网络提早减少，从而影响事务的整体性能
3PC 容易受到系统故障的影响，可能导致数据不统一

3. 基于后盾队列的异步解决(Using Queues to Process Asynchronously in the Background)

基于音讯代理或音讯队列在后盾异步解决工作，主线程能够继续执行其余工作，而不会被耗时的工作阻塞。
工作被推送到队列中，独自的工作过程从队列中读取并执行工作。
通常包含以下步骤:
- 创立队列，将异步工作存储在队列中。
- 向队列中增加须要异步解决的工作，如发送电子邮件、解决文件或生成报告。
- 创立工作过程或线程，负责执行队列中的工作，工作过程 / 线程能够应用多过程或线程库创立。
- 一旦工作过程 / 线程被创立，应用 start() 办法。
- 工作过程 / 线程将从队列中检索工作，并在后盾异步执行，重复执行直到队列中没有更多的工作。
- 一旦工作实现，工作过程 / 线程将标记该工作已实现，并将后果存储在独自的输入队列中。
- 应用 get() 办法从输入队列中检索后果。
- 应用 try-except 块解决工作执行过程中可能产生的任何异样。
- 所有工作一旦实现，并且可能检索到后果，就用 join() 办法阻塞工作线程。
- 通过敞开队列并终止工作过程 / 线程来清理资源

import multiprocessing

# Create a Queue
task_queue = multiprocessing.Queue()
result_queue = multiprocessing.Queue()

# Define a task function
def do_task(task):
    # Code to perform the task
    result = task.upper()
    # Store the result in the output queue
    result_queue.put(result)

# Add tasks to the queue
tasks = ['task1', 'task2', 'task3']
for task in tasks:
    task_queue.put(task)

# Create workers
num_workers = multiprocessing.cpu_count()
workers = [multiprocessing.Process(target=do_task, args=(task_queue,)) for i in range(num_workers)]

# Start workers
for worker in workers:
    worker.start()

# Retrieve tasks from the queue
while not task_queue.empty():
    task = task_queue.get()

# Process completed tasks
while not result_queue.empty():
    result = result_queue.get()
    print(result)

# Join workers
for worker in workers:
    worker.join()

# Clean up
task_queue.close()
result_queue.close()

示例代码

通过 Python 的 multiprocessing 模块创立工作队列和输入队列。
工作队列用于存储须要异步执行的工作，而输入队列用于存储工作的后果。
通过 put() 办法将工作增加到工作队列。
通过 cpu_count() 办法确定工作过程 / 线程数量。
通过 multiprocessing 模块的 Process 类创立工作过程对象，并将 do_tasks() 函数调配为工作过程对象的指标。
通过 start() 办法启动工作过程，通过 get() 办法从工作队列中检索工作，直到队列中没有更多任务为止。
检索到工作后，do_task()函数将对其进行解决，并应用 put() 办法将后果存储在输入队列中。
实现所有工作后，应用 get() 办法从输入队列检索后果。
通过 join() 办法阻塞工作过程，并应用 close() 办法敞开工作和后果队列。

长处

队列能够反对同时执行多个工作
队列提供了处理错误和重试的办法，确保工作在失败时不会被抛弃
队列有助于解耦零碎的不同组件

毛病

因为队列须要额定组件(如音讯代理或音讯队列)，减少了额定的复杂性
在音讯序列化和网络通信方面引入额定开销，可能会影响性能

实用场景

须要解决大量工作并对工作进行异步解决的零碎
图像和视频解决、领取和电子邮件发送等工作的异步解决
能够用来实现微服务之间的通信
可用于批量解决大量数据，如数据荡涤和转换

挑战

保护队列中音讯的程序可能会有问题
音讯被传递到正确的工作过程 / 线程，没有音讯失落或反复音讯
大量音讯的解决具备挑战性

参考文献

Implementing Retry In Kafka Consumer

retry logic blocks the main consumer while its waiting for the retry in spring

Main Difference between 2PC and 3PC Protocols

Three-Phase Commit Protocol

Distributed DBMS – Commit Protocols

COMMIT Protocol in DBMS

Asynchronous Task Processing in Cloud

Asynchronous Task Queues

Asynchronous Messaging, Part 3: Backend Service

Async and Background Processing

Queue instruction-run an activity asynchronously

你好，我是俞凡，在 Motorola 做过研发，当初在 Mavenir 做技术工作，对通信、网络、后端架构、云原生、DevOps、CICD、区块链、AI 等技术始终保持着浓重的趣味，平时喜爱浏览、思考，置信继续学习、一生成长，欢送一起交流学习。\
微信公众号：DeepNoMind

本文由 mdnice 多平台公布