运行您自己的事务

使用 executableQuery() 查询数据库时,驱动程序会自动创建一个事务。事务是工作的单元,要么完全提交,要么在失败时回滚。您可以在单个查询中包含多个 Cypher 语句,例如当按顺序使用 MATCHCREATE更新数据库时,但您不能有多个查询并在它们之间交织一些客户端逻辑。

对于这些更高级的使用案例,驱动程序提供了函数来完全控制事务生命周期。这些称为托管事务,您可以将它们视为解开 executableQuery() 的流程并能够在更多位置指定其所需行为的一种方式。

创建会话

在运行事务之前,您需要获取一个会话。会话充当驱动程序和服务器之间的具体查询通道,并确保因果一致性得到执行。

会话是使用方法Driver.session()创建的。使用可选参数更改会话的配置,例如目标数据库。有关其他配置参数,请参阅会话配置

// import org.neo4j.driver.SessionConfig

try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {
    // session usage
}

会话创建是一个轻量级操作,因此可以在没有显著成本的情况下创建和销毁会话。当您完成会话时,请务必关闭会话

会话不是线程安全的:您可以在线程之间共享主要的 Driver 对象,但请确保每个线程都创建自己的会话。

运行托管事务

一个事务可以包含任意数量的查询。由于 Neo4j 符合ACID,因此事务内的查询要么全部执行,要么根本不执行:您无法获得部分事务成功而另一部分失败的情况。使用事务将相关查询组合在一起,这些查询协同工作以实现单个逻辑数据库操作。

托管事务是使用方法Session.executeRead()Session.executeWrite()创建的,具体取决于您是想从数据库中检索数据还是更改数据。这两种方法都接受一个事务函数回调,该回调负责实际执行查询和处理结果。

检索名称以 Al 开头的人员。
// import java.util.Map
// import org.neo4j.driver.SessionConfig

try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {  (1)
    var people = session.executeRead(tx -> {  (2)
        var result = tx.run("""
        MATCH (p:Person) WHERE p.name STARTS WITH $filter  (3)
        RETURN p.name AS name ORDER BY name
        """, Map.of("filter", "Al"));
        return result.list();  // return a list of Record objects (4)
    });
    people.forEach(person -> {
        System.out.println(person);
    });
    // further tx.run() calls will execute within the same transaction
}
1 创建一个会话。单个会话可以是多个查询的容器。除非使用 try 结构作为资源创建,否则请记住在完成时关闭它。
2 .executeRead()(或 .executeWrite())方法是事务的入口点。它接收对事务函数的回调,该函数负责运行查询。
3 使用方法tx.run()执行查询。您可以提供查询参数的映射作为第二个参数。每个运行的查询都返回一个Result对象。
4 处理结果,使用 Result 上的任何方法。.list() 方法将所有记录检索到列表中。

不要将参数直接硬编码或连接到查询中。出于性能和安全原因,请改用查询参数

事务函数绝不应直接返回 Result 对象。相反,始终以某种方式处理结果。在事务函数中,return 语句会导致事务提交,而如果引发异常,则事务会自动回滚。

.executeRead().executeWrite() 方法已取代 .readTransaction().writeTransaction(),后者在 5.x 版本中已弃用,将在 6.0 版本中删除。
具有多个查询、客户端逻辑和潜在回滚的事务
package demo;

import java.util.Map;
import java.util.List;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

import org.neo4j.driver.AuthTokens;
import org.neo4j.driver.GraphDatabase;
import org.neo4j.driver.QueryConfig;
import org.neo4j.driver.Record;
import org.neo4j.driver.RoutingControl;
import org.neo4j.driver.SessionConfig;
import org.neo4j.driver.TransactionContext;
import org.neo4j.driver.exceptions.NoSuchRecordException;

public class App {

    // Create & employ 100 people to 10 different organizations
    public static void main(String... args) {

        final String dbUri = "<URI for Neo4j database>";
        final String dbUser = "<Username>";
        final String dbPassword = "<Password>";

        try (var driver = GraphDatabase.driver(dbUri, AuthTokens.basic(dbUser, dbPassword))) {
            try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {
                for (int i=0; i<100; i++) {
                    String name = String.format("Thor%d", i);

                    try {
                        String orgId = session.executeWrite(tx -> employPersonTx(tx, name));
                        System.out.printf("User %s added to organization %s.%n", name, orgId);
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        }
    }

    static String employPersonTx(TransactionContext tx, String name) {
        final int employeeThreshold = 10;

        // Create new Person node with given name, if not exists already
        tx.run("MERGE (p:Person {name: $name})", Map.of("name", name));

        // Obtain most recent organization ID and the number of people linked to it
        var result = tx.run("""
            MATCH (o:Organization)
            RETURN o.id AS id, COUNT{(p:Person)-[r:WORKS_FOR]->(o)} AS employeesN
            ORDER BY o.createdDate DESC
            LIMIT 1
            """);

        Record org = null;
        String orgId = null;
        int employeesN = 0;
        try {
            org = result.single();
            orgId = org.get("id").asString();
            employeesN = org.get("employeesN").asInt();
        } catch (NoSuchRecordException e) {
            // The query is guaranteed to return <= 1 results, so if.single() throws, it means there's none.
            // If no organization exists, create one and add Person to it
            orgId = createOrganization(tx);
            System.out.printf("No orgs available, created %s.%n", orgId);
        }

        // If org does not have too many employees, add this Person to it
        if (employeesN < employeeThreshold) {
            addPersonToOrganization(tx, name, orgId);
            // If the above throws, the transaction will roll back
            // -> not even Person is created!

        // Otherwise, create a new Organization and link Person to it
        } else {
            orgId = createOrganization(tx);
            System.out.printf("Latest org is full, created %s.%n", orgId);
            addPersonToOrganization(tx, name, orgId);
            // If any of the above throws, the transaction will roll back
            // -> not even Person is created!
        }

        return orgId;  // Organization ID to which the new Person ends up in
    }

    static String createOrganization(TransactionContext tx) {
        var result = tx.run("""
            CREATE (o:Organization {id: randomuuid(), createdDate: datetime()})
            RETURN o.id AS id
        """);
        var org = result.single();
        var orgId = org.get("id").asString();
        return orgId;
    }

    static void addPersonToOrganization(TransactionContext tx, String personName, String orgId) {
        tx.run("""
            MATCH (o:Organization {id: $orgId})
            MATCH (p:Person {name: $name})
            MERGE (p)-[:WORKS_FOR]->(o)
            """, Map.of("orgId", orgId, "name", personName)
        );
    }
}

如果事务由于驱动程序认为是瞬态的原因而失败,它会自动重试运行事务函数(并以指数增长的延迟)。因此,事务函数必须是幂等的(即,当运行多次时,它们应该产生相同的效果),因为您事先不知道它们将要执行多少次。在实践中,这意味着您不应编辑或依赖全局变量,例如。请注意,尽管事务函数可能会执行多次,但其中的查询始终只会运行一次。

一个会话可以链接多个事务,但在任何给定时间,一个会话中只能有一个事务处于活动状态。要维护多个并发事务,请使用多个并发会话。

传递给 .executeRead().executeWrite() 的事务函数回调可以返回任何内容,因为返回类型是 Java 泛型。这也意味着它们不能返回 void,因为这不是 Object 的实例。
如果您不希望从事务函数中返回任何内容,您可以

  1. 使用 .executeWriteWithoutResult(),它支持返回 void(并且仅返回 void);

  2. 使用 .executeRead()/.executeWrite()并在事务函数中返回 null

运行显式事务

您可以通过使用方法Session.beginTransaction()手动开始一个事务来实现对事务的完全控制,该方法返回一个Transaction对象。然后,您可以使用 Transaction.run() 方法在显式事务中运行查询。

try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {
    try (Transaction tx = session.beginTransaction()) {
        // use tx.run() to run queries
        //     tx.commit() to commit the transaction
        //     tx.rollback() to rollback the transaction
    }
}

显式事务可以使用Transaction.commit()提交,也可以使用Transaction.rollback()回滚。如果没有采取任何显式操作,驱动程序将在其生命周期结束时自动回滚事务。

显式事务最适用于需要跨多个函数分发 Cypher 执行以进行相同事务的应用程序,或需要在单个事务中运行多个查询但无需托管事务提供的自动重试的应用程序。

涉及外部 API 的显式事务示例
package demo;

import java.util.Map;
import java.util.List;
import java.util.Arrays;

import org.neo4j.driver.AuthTokens;
import org.neo4j.driver.Driver;
import org.neo4j.driver.GraphDatabase;
import org.neo4j.driver.QueryConfig;
import org.neo4j.driver.Record;
import org.neo4j.driver.SessionConfig;
import org.neo4j.driver.Transaction;

public class App {

    public static void main(String... args) {
        final String dbUri = "<URI for Neo4j database>";
        final String dbUser = "<Username>";
        final String dbPassword = "<Password>";

        try (var driver = GraphDatabase.driver(dbUri, AuthTokens.basic(dbUser, dbPassword))) {
            driver.verifyConnectivity();

            String customerId = createCustomer(driver);
            int otherBankId = 42;
            transferToOtherBank(driver, customerId, otherBankId, 999);
        }
    }

    static String createCustomer(Driver driver) {
        var result = driver.executableQuery("""
            MERGE (c:Customer {id: randomUUID(), balance: 1000})
            RETURN c.id AS id
            """)
            .withConfig(QueryConfig.builder().withDatabase("neo4j").build())
            .execute();
        return result.records().get(0).get("id").asString();
    }

    static void transferToOtherBank(Driver driver, String customerId, int otherBankId, float amount) {
        try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {
            try (var tx = session.beginTransaction()) {
                if (! customerBalanceCheck(tx, customerId, amount)) {
                    System.out.printf("Customer %s doesn't have enough funds.%n", customerId);
                    return;  // give up
                }

                otherBankTransferApi(customerId, otherBankId, amount);
                // Now the money has been transferred => can't rollback anymore
                // (cannot rollback external services interactions)

                try {
                    decreaseCustomerBalance(tx, customerId, amount);
                    tx.commit();
                    System.out.printf("Transferred %f to %s.%n", amount, customerId);
                } catch (Exception e) {
                    requestInspection(customerId, otherBankId, amount, e);
                    throw new RuntimeException(e.getMessage());
                }
            }
        }
    }

    static boolean customerBalanceCheck(Transaction tx, String customerId, float amount) {
        var result = tx.run("""
            MATCH (c:Customer {id: $id})
            RETURN c.balance >= $amount AS sufficient
            """, Map.of("id", customerId, "amount", amount));
        var record = result.single();
        return record.get("sufficient").asBoolean();
    }

    static void otherBankTransferApi(String customerId, int otherBankId, float amount) {
        // make some API call to other bank
    }

    static void decreaseCustomerBalance(Transaction tx, String customerId, float amount) {
        tx.run("""
            MATCH (c:Customer {id: $id})
            SET c.balance = c.balance - $amount
            """, Map.of("id", customerId, "amount", amount));
    }

    static void requestInspection(String customerId, int otherBankId, float amount, Exception e) {
        // manual cleanup required; log this or similar
        System.out.printf("WARNING: transaction rolled back due to exception: %s.%n", e.getMessage());
        System.out.printf("customerId: %s, otherBankId: %d, amount: %f.%n", customerId, otherBankId, amount);
    }
}

处理查询结果

驱动程序的查询输出是一个Result对象,它将 Cypher 结果封装在一个丰富的数据结构中,需要在客户端进行一些解析。需要注意两个主要要点

  • 服务器不会立即获取并返回所有结果记录。相反,结果以惰性流的形式返回。具体来说,当驱动程序从服务器接收一些记录时,它们最初会被缓冲到后台队列中。记录会保留在缓冲区中,直到被应用程序消费,此时它们会被从缓冲区中移除。当没有更多记录可用时,结果就被认为是耗尽的。

  • 结果充当一个游标。这意味着无法从流中检索之前的记录,除非您将其保存在辅助数据结构中。

下面的动画演示了一个单个查询的执行路径:它展示了驱动程序如何处理结果记录以及应用程序应该如何处理结果。

处理结果最简单的方法是调用其上的.list()方法,该方法会生成一个Record对象列表。否则,Result对象实现了多个用于处理记录的方法。下面列出了最常用的方法。

方法 描述

list() List<Record>

将结果的剩余部分作为列表返回。

single() Record

返回下一个且唯一的剩余记录。调用此方法始终会耗尽结果。如果可用记录多于(或少于)一个,则会引发NoSuchRecordException异常。

next() Record

返回结果中的下一条记录。如果不再有记录可用,则抛出NoSuchRecordException异常。

hasNext() boolean

结果迭代器是否还有下一条记录可以移动到。

peek() Record

不消费结果中的下一条记录并将其返回。这会将记录保留在缓冲区中以供进一步处理。

consume() ResultSummary

返回查询的结果摘要。它会耗尽结果,因此只有在数据处理完成后才能调用。

有关Result方法的完整列表,请参阅API 文档 → Result

Record对象中的属性嵌套在Value对象中。要提取并将它们转换为相应的Java类型,请使用.as<type>()(例如.asString()asInt()等)。例如,如果来自数据库的name属性是字符串,则record.get("name").asString()会将属性值作为String对象返回。

有关更多信息,请参阅数据类型和与Cypher类型的映射

会话配置

数据库选择

建议始终通过.withDatabase("<dbName>")方法显式指定数据库,即使在单数据库实例上也是如此。这使驱动程序能够更高效地工作,因为它节省了到服务器的网络往返以解析主数据库。如果未提供数据库,则使用Neo4j实例设置中设置的默认数据库

// import org.neo4j.driver.SessionConfig;

var session = driver.session(SessionConfig.builder()
    .withDatabase("neo4j").build());
通过配置方法指定数据库比使用USE Cypher子句更可取。如果服务器在集群上运行,则使用USE的查询需要启用服务器端路由。查询的执行时间也可能更长,因为它们可能无法在第一次尝试时到达正确的集群成员,并且需要路由到包含请求数据库的一个成员。

请求路由

在集群环境中,所有会话都以写入模式打开,将其路由到领导者。您可以通过调用方法.withRouting(RoutingControl.READ)来更改此设置。请注意,.executeRead().executeWrite()会自动覆盖会话的默认访问模式。

// import org.neo4j.driver.SessionConfig;
// import org.neo4j.driver.AccessMode;

var session = driver.session(SessionConfig.builder()
    .withDatabase("neo4j")
    .withDefaultAccessMode(AccessMode.READ)
    .build());

尽管在读取模式下执行写入查询可能会导致运行时错误,但您不应该依赖此功能进行访问控制。这两种模式之间的区别在于,读取事务被路由到集群的任何节点,而写入事务则被定向到领导者。换句话说,无法保证以读取模式提交的写入查询会被拒绝。

类似的说明也适用于.executeRead().executeWrite()方法。

以其他用户身份运行查询(模拟)

您可以使用.withImpersonatedUser("<username>")方法在其他用户的安全上下文中执行查询,指定要模拟的用户名称。要使此方法生效,创建Driver的用户需要具有相应的权限。模拟用户比创建新的Driver对象更便宜。

// import org.neo4j.driver.SessionConfig;
// import org.neo4j.driver.RoutingControl;

var session = driver.session(SessionConfig.builder()
    .withDatabase("neo4j")
    .withImpersonatedUser("somebodyElse")
    .build());

模拟用户时,查询将在模拟用户的完整安全上下文中运行,而不是已认证的用户(即主数据库、权限等)。

事务配置

您可以通过提供一个TransactionConfig对象作为.executeRead().executeWrite().beginTransaction()的(可选)第二个参数来进一步控制事务。使用它来指定

  • 事务超时。运行时间过长的事务将被服务器终止。默认值在服务器端设置。最小值为 1 毫秒。

  • 附加到事务的元数据映射。这些元数据将在服务器的query.log中记录,并且在SHOW TRANSACTIONS Cypher命令的输出中可见。使用它来标记事务。

// import java.time.Duration
// import org.neo4j.driver.SessionConfig
// import org.neo4j.driver.TransactionConfig

try (var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build())) {
    var people = session.executeRead(tx -> {
        var result = tx.run("MATCH (p:Person) RETURN p");
        return result.list();  // return a list of Record objects
    }, TransactionConfig.builder()
        .withTimeout(Duration.ofSeconds(5))
        .withMetadata(Map.of("appName", "peopleTracker"))
        .build()
    );
    people.forEach(person -> System.out.println(person));
}

关闭会话

每个连接池都有有限数量的会话,因此,如果您打开会话而不关闭它们,您的应用程序可能会用完这些会话。因此,建议使用try-with-resources语句创建会话,该语句会在应用程序完成使用会话时自动关闭它们。关闭会话后,它将返回到连接池以供以后重用。

如果您没有使用try将会话作为资源打开,请记住在完成使用它们后调用.close()方法。

var session = driver.session(SessionConfig.builder().withDatabase("neo4j").build());

// session usage

session.close();

术语表

LTS

长期支持版本是保证支持若干年的版本。Neo4j 4.4 是 LTS 版本,Neo4j 5 也将有一个 LTS 版本。

Aura

Aura 是 Neo4j 的完全托管云服务。它提供免费和付费计划。

Cypher

Cypher 是 Neo4j 的图查询语言,允许您从数据库中检索数据。它类似于 SQL,但用于图数据。

APOC

Awesome Procedures On Cypher (APOC) 是一个包含(许多)函数的库,这些函数无法用 Cypher 本身轻松表达。

Bolt

Bolt 是用于 Neo4j 实例和驱动程序之间交互的协议。默认情况下,它侦听端口 7687。

ACID

原子性、一致性、隔离性和持久性 (ACID) 是保证数据库事务可靠处理的特性。符合 ACID 的 DBMS 确保数据库中的数据在发生故障时仍然准确且一致。

最终一致性

如果数据库保证所有集群成员将在某个时间点存储数据的最新版本,则该数据库最终一致。

因果一致性

如果读取和写入查询以相同的顺序由集群的每个成员看到,则该数据库是因果一致的。这比最终一致性更强。

NULL

空标记不是一种类型,而是值不存在的占位符。有关更多信息,请参阅Cypher → 使用null

事务

事务是一项工作单元,要么完全提交,要么在发生故障时回滚。例如银行转账:它涉及多个步骤,但它们必须全部成功或回滚,以避免从一个账户中扣除资金但未添加到另一个账户中。

背压

背压是与数据流相反的力。它确保客户端不会被超出其处理能力的数据速度所淹没。

事务函数

事务函数是由executeReadexecuteWrite调用执行的回调。如果服务器发生故障,驱动程序会自动重新执行回调。

Driver

一个Driver对象包含建立与Neo4j数据库连接所需的详细信息。