2.5 事务与 ACID

文档摘要

2.5 事务与 ACID 2.5 事务与 ACID：Neo4j 数据一致性的基石 2.5.1 事务的概念与重要性在数据库系统中，事务是指作为单个逻辑工作单元执行的一系列操作。这些操作要么全部成功执行（提交 - Commit），要么全部不执行（回滚 - Rollback）。事务的主要目的是保证数据库在并发操作和系统故障的情况下，仍然能够保持数据的一致性和完整性。在 Neo4j 中，事务的概念同样至关重要。无论是创建新的节点和关系，还是更新或删除现有数据，都应该在事务的保护下进行。事务确保了在执行复杂图操作时，即使发生错误或系统崩溃，数据库也能回退到操作前的状态，避免数据损坏或不一致。为什么事务如此重要？数据一致性：事务确保数据从一个一致的状态转移到另一个一致的状态。

2.5 事务与 ACID

2.5 事务与 ACID：Neo4j 数据一致性的基石

2.5.1 事务的概念与重要性

在数据库系统中，事务是指作为单个逻辑工作单元执行的一系列操作。这些操作要么全部成功执行（提交 - Commit），要么全部不执行（回滚 - Rollback）。事务的主要目的是保证数据库在并发操作和系统故障的情况下，仍然能够保持数据的一致性和完整性。

在 Neo4j 中，事务的概念同样至关重要。无论是创建新的节点和关系，还是更新或删除现有数据，都应该在事务的保护下进行。事务确保了在执行复杂图操作时，即使发生错误或系统崩溃，数据库也能回退到操作前的状态，避免数据损坏或不一致。

为什么事务如此重要？

数据一致性： 事务确保数据从一个一致的状态转移到另一个一致的状态。即使在多个操作同时发生的情况下，事务也能保证数据在逻辑上的完整性和正确性。
并发控制： 在多用户或多线程环境中，事务可以隔离并发操作，防止数据互相干扰，保证每个用户都能看到一致的数据视图。
错误恢复： 当事务执行过程中发生错误时，可以回滚事务，撤销已经执行的操作，使数据库恢复到事务开始前的状态，增强系统的健壮性。
业务逻辑完整性： 许多业务操作都需要多个步骤才能完成，事务可以将这些步骤捆绑成一个原子操作，保证业务逻辑的完整性。

2.5.2 ACID 属性详解

ACID 是一组数据库事务的特性，是保证数据可靠性的核心原则。ACID 分别代表：

原子性 (Atomicity)
一致性 (Consistency)
隔离性 (Isolation)
持久性 (Durability)

接下来，我们将详细解释每个 ACID 属性在 Neo4j 中的含义和实现方式。

2.5.2.1 原子性 (Atomicity)

原子性 指的是事务是一个不可分割的工作单元，事务中的所有操作要么全部成功提交，要么全部失败回滚。不存在部分成功、部分失败的情况。就像原子是不可再分的最小粒子一样，事务也是数据库操作的最小不可分割单元。

在 Neo4j 中，原子性是如何实现的？

Neo4j 使用事务日志（Transaction Log） 来实现原子性。当一个事务开始时，Neo4j 会记录事务中执行的所有操作到事务日志中。如果事务成功提交，日志中的操作将被应用到数据库中。如果事务失败或被回滚，Neo4j 会根据事务日志撤销已经执行的操作，使数据库恢复到事务开始前的状态。

代码实践：原子性演示

我们使用 Python Neo4j 驱动来演示原子性。假设我们要创建一个用户节点和一个地址节点，并建立 LIVES_AT 关系。如果在创建地址节点时发生错误，整个事务应该回滚，用户节点和关系都不应该被创建。


from neo4j import GraphDatabase
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password"
driver = GraphDatabase.driver(uri, auth=(username, password))
def create_user_and_address(tx, user_name, address_city):
    try:
        # 尝试创建一个用户节点
        user_node = tx.run("CREATE (u:User {name: $user_name}) RETURN u", user_name=user_name).single()
        print(f"成功创建用户节点: {user_node}")
        # 模拟错误：假设地址城市为空时会抛出异常
        if not address_city:
            raise ValueError("地址城市不能为空")
        # 尝试创建一个地址节点并建立关系
        address_node = tx.run("CREATE (a:Address {city: $address_city}) RETURN a", address_city=address_city).single()
        tx.run("MATCH (u:User {name: $user_name}), (a:Address {city: $address_city}) CREATE (u)-[:LIVES_AT]->(a)",
               user_name=user_name, address_city=address_city)
        print(f"成功创建地址节点并建立关系: {address_node}")
        return "操作成功"
    except ValueError as e:
        print(f"发生错误: {e}")
        raise e # 显式抛出异常，触发事务回滚
with driver.session() as session:
    try:
        result = session.execute_write(create_user_and_address, "Alice", "") # 地址城市为空，模拟错误
        print(f"事务结果: {result}")
    except ValueError:
        print("事务已回滚")
    # 验证数据是否回滚，用户节点和关系不应该被创建
    query_user = session.run("MATCH (u:User {name: 'Alice'}) RETURN u").peek()
    query_address = session.run("MATCH (a:Address) WHERE a.city IS NULL RETURN a").peek() # 假设没有其他空城市地址
    if query_user:
        print("用户节点仍然存在，原子性失败！") # 不应该执行到这里
    else:
        print("用户节点不存在，原子性验证成功。")
    if query_address:
        print("地址节点仍然存在，原子性失败！") # 不应该执行到这里
    else:
        print("地址节点不存在，原子性验证成功。")
driver.close()

代码详解:

我们使用 session.execute_write() 方法来执行写操作事务。
create_user_and_address 函数定义了事务中的操作：创建用户节点、创建地址节点、建立关系。
我们故意将 address_city 设置为空字符串，模拟在创建地址节点时发生 ValueError 异常。
在 try...except 块中，我们捕获 ValueError 异常，并显式地 raise e，这会触发 Neo4j 驱动回滚当前事务。
最后，我们查询数据库，验证用户节点和地址节点是否被创建。由于事务回滚，我们应该看不到任何新创建的节点。

mermaid 图示：原子性

图示解释:

事务从 "事务开始" 节点开始。
"操作 1" 和 "操作 2" 代表事务中的两个步骤。
如果 "操作 2" 失败（例如抛出异常），事务会进入 "事务回滚" 状态。
"事务回滚" 会使数据库恢复到事务开始前的状态，之前的 "操作 1" 的效果也会被撤销。
如果所有操作都成功，事务会进入 "事务提交" 状态。
"事务提交" 会将所有更改持久化到数据库中。

2.5.2.2 一致性 (Consistency)

一致性 指的是事务必须保证数据库从一个一致性状态转换到另一个一致性状态。一致性状态是指数据库满足所有预定义的规则和约束的状态，例如数据类型约束、唯一性约束、外键约束、应用层定义的业务规则等。

在 Neo4j 中，一致性是如何实现的？

Neo4j 通过以下机制来保证一致性：

模式约束 (Schema Constraints)： Neo4j 允许用户定义模式约束，例如节点属性的类型约束、唯一性约束等。事务在提交前会检查是否违反了这些约束，如果违反，事务会回滚。
数据完整性规则 (Data Integrity Rules)： Neo4j 内部维护着图数据库的数据完整性规则，例如节点和关系必须符合图模型的定义，关系必须连接到节点等。事务操作必须符合这些规则，否则会被拒绝。
应用层逻辑 (Application Logic)： 一致性也依赖于应用层正确的业务逻辑。事务应该按照业务规则来操作数据，保证数据的逻辑一致性。

代码实践：一致性演示

假设我们有一个用户节点，我们需要为其设置一个唯一的邮箱属性。我们创建一个唯一性约束，然后尝试创建两个邮箱相同的用户，验证一致性约束是否生效。

首先，在 Neo4j Browser 或 Cypher Shell 中创建唯一性约束：


CREATE CONSTRAINT UserEmailUnique FOR (u:User) REQUIRE u.email IS UNIQUE

然后，使用 Python 驱动演示：


from neo4j import GraphDatabase
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password"
driver = GraphDatabase.driver(uri, auth=(username, password))
def create_user_with_email(tx, user_name, user_email):
    try:
        tx.run("CREATE (u:User {name: $user_name, email: $user_email}) RETURN u",
               user_name=user_name, user_email=user_email)
        return "用户创建成功"
    except Exception as e: # 捕获更广泛的异常，包括约束违反
        print(f"创建用户失败: {e}")
        raise e # 显式抛出异常，触发事务回滚
with driver.session() as session:
    email = "test@example.com"
    # 第一次创建用户，应该成功
    try:
        result1 = session.execute_write(create_user_with_email, "User1", email)
        print(f"事务 1 结果: {result1}")
    except Exception:
        print("事务 1 已回滚 (不应该发生)")
    # 第二次创建用户，使用相同的邮箱，应该违反唯一性约束，事务回滚
    try:
        result2 = session.execute_write(create_user_with_email, "User2", email)
        print(f"事务 2 结果: {result2}") # 不应该执行到这里
    except Exception:
        print("事务 2 已回滚 (违反唯一性约束)")
    # 验证数据库状态，应该只有一个邮箱为 test@example.com 的用户
    count = session.run("MATCH (u:User {email: $email}) RETURN count(u)", email=email).single()[0]
    print(f"邮箱为 {email} 的用户数量: {count}") # 应该输出 1
driver.close()

代码详解:

我们首先在 Neo4j 中创建了 User.email 属性的唯一性约束。
create_user_with_email 函数尝试创建带有指定邮箱的用户节点。
我们第一次创建用户 "User1" 时，应该成功。
第二次创建用户 "User2" 时，使用相同的邮箱 test@example.com，这会违反唯一性约束。Neo4j 会检测到约束 violation 并拒绝提交事务，事务会回滚。
最后，我们查询数据库中邮箱为 test@example.com 的用户数量，应该只有 1 个，证明一致性约束起作用了。

mermaid 图示：一致性

图示解释:

事务从 "事务开始" 节点开始。
"操作" 代表创建用户节点的操作。
如果 "操作" 违反了数据库的约束（例如唯一性约束），Neo4j 会检测到 "一致性错误"。
"一致性错误" 会导致 "事务回滚"，数据库会 "保持一致性状态"，即回滚到事务开始前的状态，违反约束的操作不会生效。
如果 "操作" 符合所有约束，事务会 "事务提交"，数据库会 "更新到新的 consistent 状态"。

2.5.2.3 隔离性 (Isolation)

隔离性 指的是并发执行的事务之间应该互相隔离，一个事务的执行不应该受到其他事务的干扰。每个事务都应该感觉像是在独立地访问数据库，即使实际上可能有多个事务同时在运行。隔离性保证了并发环境下的数据一致性。

在 Neo4j 中，隔离性是如何实现的？

Neo4j 默认提供 完全隔离 (Serializable) 级别。这意味着并发事务的执行效果应该等同于它们按照某种顺序串行执行的效果。Neo4j 使用 锁机制 (Locking Mechanisms) 来实现隔离性。当一个事务访问或修改数据时，Neo4j 会对相关数据加锁，防止其他事务同时修改或读取正在被修改的数据，从而保证隔离性。

代码实践：隔离性演示

我们模拟两个并发事务，一个事务读取用户的年龄，另一个事务同时更新用户的年龄。我们将演示在默认的 Serializable 隔离级别下，读取事务不会受到更新事务的未提交更改的影响，保证了隔离性。


import threading
import time
from neo4j import GraphDatabase
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password"
driver = GraphDatabase.driver(uri, auth=(username, password))
def setup_user():
    with driver.session() as session:
        session.run("MERGE (u:User {name: 'ConcurrentUser'}) SET u.age = 30")
def read_user_age(tx):
    age = tx.run("MATCH (u:User {name: 'ConcurrentUser'}) RETURN u.age AS age").single()["age"]
    print(f"读取事务：初始年龄为 {age}")
    time.sleep(2) # 模拟读取事务需要一些时间
    age_again = tx.run("MATCH (u:User {name: 'ConcurrentUser'}) RETURN u.age AS age").single()["age"]
    print(f"读取事务：再次读取年龄为 {age_again}") # 在 Serializable 隔离级别下，应该仍然看到初始年龄
def update_user_age(tx):
    print("更新事务：开始更新年龄")
    tx.run("MATCH (u:User {name: 'ConcurrentUser'}) SET u.age = 40")
    print("更新事务：年龄已更新为 40，但事务未提交")
    time.sleep(4) # 模拟更新事务执行时间比读取事务长
    print("更新事务：提交事务") # 更新事务提交
    return "年龄更新完成"
def run_transaction_read():
    with driver.session() as session:
        session.execute_read(read_user_age)
def run_transaction_write():
    with driver.session() as session:
        session.execute_write(update_user_age)
if __name__ == "__main__":
    setup_user() # 初始化用户数据
    thread_read = threading.Thread(target=run_transaction_read)
    thread_write = threading.Thread(target=run_transaction_write)
    thread_read.start()
    thread_write.start()
    thread_read.join()
    thread_write.join()
    # 最终验证数据库中的年龄
    with driver.session() as session:
        final_age = session.run("MATCH (u:User {name: 'ConcurrentUser'}) RETURN u.age AS age").single()["age"]
        print(f"最终用户年龄为: {final_age}") # 最终年龄应该是 40
    driver.close()

代码详解:

我们使用 Python 线程模拟两个并发事务：read_transaction 和 update_transaction。
read_user_age 函数模拟一个读取事务，它读取用户的年龄两次，中间暂停一段时间。
update_user_age 函数模拟一个更新事务，它将用户的年龄更新为 40，也暂停一段时间，然后提交事务。
我们启动两个线程并发执行这两个事务。
在 Serializable 隔离级别下，read_transaction 在第一次读取年龄后，即使 update_transaction 已经更新了年龄但尚未提交，read_transaction 再次读取年龄时仍然会看到事务开始时的年龄 (30)。只有当 update_transaction 提交后，其他事务才能看到更新后的年龄。
最后，我们验证数据库中用户的最终年龄，应该是 40。

mermaid 图示：隔离性 (Serializable)

图示解释:

"时间轴" 表示时间流逝。
"事务 1" 和 "事务 2" 并发开始。
"事务 1" 首先读取年龄，看到初始值 30。
"事务 2" 更新年龄为 40，但事务尚未提交。
由于 Serializable 隔离级别，当 "事务 1" 再次读取年龄时，它仍然 "看到年龄 30"，即使 "事务 2" 已经进行了修改。
只有当 "事务 2 提交" 后，数据库的 "年龄更新为 40"，后续事务才能看到更新后的值。

2.5.2.4 持久性 (Durability)

持久性 指的是一旦事务成功提交，其对数据库的更改应该永久保存下来，即使系统发生故障（例如断电、崩溃）也不会丢失。持久性保证了数据的可靠性和可恢复性。

在 Neo4j 中，持久性是如何实现的？

Neo4j 通过以下机制来保证持久性：

写前日志 (Write-Ahead Logging - WAL)： Neo4j 使用 WAL 机制。在事务提交之前，所有的更改都会先写入到事务日志文件中。只有当事务日志被成功写入磁盘后，事务才会被认为提交成功。
定期检查点 (Checkpoints)： Neo4j 定期将内存中的数据刷新到磁盘上的数据库文件中，并将检查点信息写入事务日志。这样，即使系统崩溃，Neo4j 也可以通过重放事务日志中检查点之后的日志记录，将数据库恢复到最近一次一致的状态。
数据持久化 (Data Persistence)： Neo4j 将数据库文件存储在磁盘上，确保数据在断电或其他系统故障后仍然存在。

代码实践：持久性演示

持久性的验证通常需要模拟系统故障，这在代码层面比较困难直接演示。但是我们可以通过一个简单的例子来展示事务提交后数据确实被持久化了。


from neo4j import GraphDatabase
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password"
driver = GraphDatabase.driver(uri, auth=(username, password))
def create_persistent_node(tx, node_name):
    tx.run("CREATE (n:PersistentNode {name: $node_name}) RETURN n", node_name=node_name)
    return "节点创建并提交"
with driver.session() as session:
    result = session.execute_write(create_persistent_node, "PersistentNode1")
    print(f"事务结果: {result}")
driver.close()
# 假设程序运行结束后，系统发生故障 (例如重启 Neo4j 服务)
# 重新连接 Neo4j 并验证数据是否仍然存在
driver_reconnect = GraphDatabase.driver(uri, auth=(username, password))
with driver_reconnect.session() as session_reconnected:
    node_exists = session_reconnected.run("MATCH (n:PersistentNode {name: 'PersistentNode1'}) RETURN n").peek()
    if node_exists:
        print("持久性验证成功：节点在重启后仍然存在")
    else:
        print("持久性验证失败：节点在重启后丢失") # 不应该发生
driver_reconnect.close()

代码详解:

create_persistent_node 函数创建一个名为 "PersistentNode1" 的节点。
我们执行事务并提交。
假设在程序运行结束后，我们模拟系统故障，例如重启 Neo4j 服务。
我们重新连接到 Neo4j 数据库。
我们查询数据库，验证之前创建的 "PersistentNode1" 节点是否仍然存在。由于持久性保证，节点应该仍然存在。

mermaid 图示：持久性

图示解释:

"事务提交" 后，更改首先 "写入事务日志 (WAL)"。
"事务日志持久化到磁盘" 是保证持久性的关键步骤。
之后，内存中的数据会被更新，并 "定期检查点" 将内存数据刷新到磁盘数据库文件。
即使发生 "系统故障"，系统恢复后，Neo4j 也可以通过 "重放事务日志" 将数据库 "恢复到提交后的状态"，保证 "数据持久存在"。

2.5.3 Neo4j 事务管理实践

在 Neo4j 中，事务管理主要通过以下方式进行：

显式事务 (Explicit Transactions)： 使用驱动程序提供的 API (例如 Python 驱动的 session.begin_transaction(), tx.commit(), tx.rollback()) 或者 Bolt 协议的 BEGIN, COMMIT, ROLLBACK 命令来显式控制事务的开始、提交和回滚。
隐式事务 (Implicit Transactions / Auto-Commit)： 默认情况下，Neo4j Cypher 查询是自动提交 (Auto-Commit) 的。这意味着每个独立的 Cypher 查询语句都会被当做一个独立的事务来执行，执行成功自动提交，执行失败自动回滚。对于简单的单语句操作，自动提交事务非常方便。
过程 (Procedures) 中的事务控制： 在 Neo4j 过程 (Procedures) 中，可以更灵活地控制事务的边界，实现更复杂的事务逻辑。

代码实践：显式事务控制

使用 Python 驱动演示显式事务控制：


from neo4j import GraphDatabase
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password"
driver = GraphDatabase.driver(uri, auth=(username, password))
def explicit_transaction_example(tx):
    try:
        tx.run("CREATE (n:TxNode {name: 'TxNode1'})") # 操作 1
        tx.run("CREATE (n:TxNode {name: 'TxNode2'})") # 操作 2
        # 假设某些条件满足，决定提交事务
        commit_transaction = True
        if commit_transaction:
            tx.commit()
            print("显式事务已提交")
        else:
            tx.rollback()
            print("显式事务已回滚")
    except Exception as e:
        tx.rollback()
        print(f"显式事务发生错误已回滚: {e}")
with driver.session() as session:
    session.execute_transaction(explicit_transaction_example) # 使用 execute_transaction 执行显式事务
# 验证节点是否被创建 (假设事务已提交)
with driver.session() as session_verify:
    count = session_verify.run("MATCH (n:TxNode) RETURN count(n)").single()[0]
    print(f"TxNode 节点数量: {count}") # 应该输出 2 (如果事务提交)
driver.close()

代码详解:

我们使用 session.execute_transaction() 方法来执行显式事务。
explicit_transaction_example 函数接收一个事务对象 tx 作为参数。
在函数内部，我们可以执行多个 Cypher 查询操作 (操作 1, 操作 2)。
通过 tx.commit() 显式提交事务，或通过 tx.rollback() 显式回滚事务。
如果函数中发生异常，我们可以在 except 块中调用 tx.rollback() 来回滚事务。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/212-neo4j%E5%9B%BE%E6%95%B0%E6%8D%AE%E5%BA%93%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流