会话 API

本章描述以下内容

简单会话
异步会话
反应式会话
配置

简单会话

可用版本

.NET 驱动程序
Go 驱动程序
Java 驱动程序
Python 驱动程序

简单会话为 Cypher 执行提供了一个“经典”的**阻塞式** API。通常，简单会话提供最简单的编程风格，因为 API 调用以严格的顺序方式执行。

生命周期

会话生命周期从会话构造到会话关闭。在支持它们的语言中，简单会话通常在上下文块内作用域；这确保它们被正确关闭，并且任何底层连接都被释放并且不会泄漏。

示例 1. 会话构造和关闭

using (var session = driver.Session(...)) {
    // transactions go here
  }

session := driver.NewSession(...)
defer session.Close()
// transactions go here

try (Session session = driver.session(...)) {
    // transactions go here
}

with driver.session(...) as session:
    // transactions go here

**会话可以通过多种不同的方式进行配置**。这通过在会话构造函数中提供配置来实现。有关更多详细信息，请参阅会话配置。

事务函数

事务函数用于包含事务性工作单元。这种形式的事务需要最少的样板代码，并允许数据库查询和应用程序逻辑清晰分离。

事务函数也很受欢迎，因为它们封装了重试逻辑，并且在将服务器的单个实例交换为集群时提供了最大程度的灵活性。

事务函数可以作为读取或写入操作调用。此选择会将事务路由到集群环境中的适当服务器。如果您在单实例环境中操作，则此路由没有影响。如果您以后选择采用集群环境，它确实为您提供了灵活性。

在编写事务函数**之前，务必确保其设计为幂等的**。这是因为如果初始运行失败，则可能会多次执行函数。

在事务函数内获得的任何查询结果都应在该函数内使用，因为超出作用域时无法正确管理连接相关的资源。为此，事务函数可以返回值，但这些值应该是派生值而不是原始结果。

事务函数是包含事务性工作单元的推荐形式.

**当事务失败时，将调用驱动程序重试逻辑**。对于某些故障情况，可以立即对另一台服务器重试事务。这些情况包括连接问题、服务器角色更改（例如领导者选举）和瞬态错误。

示例 2. 事务函数

public void AddPerson(string name)
{
    using (var session = Driver.Session())
    {
        session.WriteTransaction(tx => tx.Run("CREATE (a:Person {name: $name})", new {name}));
    }
}

func addPersonInTxFunc(driver neo4j.Driver, name string) error {
	session := driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
	defer session.Close()

	_, err := session.WriteTransaction(func(tx neo4j.Transaction) (interface{}, error) {
		result, err := tx.Run("CREATE (a:Person {name: $name})", map[string]interface{}{"name": name})
		if err != nil {
			return nil, err
		}

		return result.Consume()
	})

	return err
}

public void addPerson( final String name )
{
    try ( Session session = driver.session() )
    {
        session.writeTransaction( tx -> {
            tx.run( "CREATE (a:Person {name: $name})", parameters( "name", name ) );
            return 1;
        } );
    }
}

from neo4j import unit_of_work

@unit_of_work(timeout=5)
def create_person(tx, name):
    return tx.run("CREATE (a:Person {name: $name}) RETURN id(a)", name=name).single().value()


def add_person(driver, name):
    with driver.session() as session:
        return session.write_transaction(create_person, name)

自动提交事务

**自动提交事务**是一种**基本但有限的事务形式**。此类事务仅包含一个 Cypher 查询，并且在失败时不会自动重放。因此，任何错误情况都必须由客户端应用程序本身处理。

自动提交事务**旨在用于简单的用例**，例如学习**Cypher** 或编写一次性脚本时。

不建议在生产环境中使用自动提交事务。

与其他类型的 Cypher 查询不同，PERIODIC COMMIT 查询不参与因果链。

因此，从驱动程序执行PERIODIC COMMIT 的唯一方法是使用自动提交事务。

请参阅Cypher 手册 → PERIODIC COMMIT 查询提示。

示例 3. 简单自动提交事务

public void AddPerson(string name)
{
    using (var session = Driver.Session())
    {
        session.Run("CREATE (a:Person {name: $name})", new {name});
    }
}

func addPersonInAutoCommitTx(driver neo4j.Driver, name string) error {
	session := driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
	defer session.Close()

	result, err := session.Run("CREATE (a:Person {name: $name})", map[string]interface{}{"name": name})
	if err != nil {
		return err
	}

	if _, err = result.Consume(); err != nil {
		return err
	}

	return nil
}

public void addPerson( String name )
{
    try ( Session session = driver.session() )
    {
        session.run( "CREATE (a:Person {name: $name})", parameters( "name", name ) );
    }
}

from neo4j import Query

def add_person(self, name):
    with self.driver.session() as session:
        session.run("CREATE (a:Person {name: $name})", name=name)

# Alternative implementation, with a one second timeout
def add_person_within_a_second(self, name):
    with self.driver.session() as session:
        session.run(Query("CREATE (a:Person {name: $name})", timeout=1.0), name=name)

使用结果

查询结果通常作为记录流使用。驱动程序提供了一种特定于语言的方式来迭代该流。

示例 4. 使用结果

public List<string> GetPeople()
{
    using (var session = Driver.Session())
    {
        return session.ReadTransaction(tx =>
        {
            var result = tx.Run("MATCH (a:Person) RETURN a.name ORDER BY a.name");
            return result.Select(record => record[0].As<string>()).ToList();
        });
    }
}

func getPeople(driver neo4j.Driver) ([]string, error) {
	session := driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
	defer session.Close()

	people, err := session.ReadTransaction(func(tx neo4j.Transaction) (interface{}, error) {
		var list []string

		result, err := tx.Run("MATCH (a:Person) RETURN a.name ORDER BY a.name", nil)
		if err != nil {
			return nil, err
		}

		for result.Next() {
			list = append(list, result.Record().Values[0].(string))
		}

		if err = result.Err(); err != nil {
			return nil, err
		}

		return list, nil
	})
	if err != nil {
		return nil, err
	}

	return people.([]string), nil
}

public List<String> getPeople()
{
    try ( Session session = driver.session() )
    {
        return session.readTransaction( tx -> {
            List<String> names = new ArrayList<>();
            Result result = tx.run( "MATCH (a:Person) RETURN a.name ORDER BY a.name" );
            while ( result.hasNext() )
            {
                names.add( result.next().get( 0 ).asString() );
            }
            return names;
        } );
    }
}

def match_person_nodes(tx):
    result = tx.run("MATCH (a:Person) RETURN a.name ORDER BY a.name")
    return [record["a.name"] for record in result]

with driver.session() as session:
    people = session.read_transaction(match_person_nodes)

保留结果

在一个会话中，**任何时候只能有一个结果流处于活动状态**。因此，如果一个查询的结果在执行另一个查询之前未完全使用，则第一个结果的其余部分将自动缓冲在结果对象中。

图 1. 结果缓冲区

此缓冲区为结果提供了一个暂存点，并将结果处理划分为**获取**（从网络移动到缓冲区）和**使用**（从缓冲区移动到应用程序）。

对于大型结果，结果缓冲区可能需要大量的内存。

因此，**建议尽可能按顺序使用结果**。

客户端应用程序可以通过显式保留结果来选择控制更高级的查询模式。当需要保存结果以供将来处理时，此类显式保留也可能很有用。驱动程序根据以下示例提供了对该过程的支持

示例 5. 保留结果以供进一步处理

public int AddEmployees(string companyName)
{
    using (var session = Driver.Session())
    {
        var persons =
            session.ReadTransaction(tx => tx.Run("MATCH (a:Person) RETURN a.name AS name").ToList());
        return persons.Sum(person => session.WriteTransaction(tx =>
        {
            tx.Run("MATCH (emp:Person {name: $person_name}) " +
                   "MERGE (com:Company {name: $company_name}) " +
                   "MERGE (emp)-[:WORKS_FOR]->(com)",
                new {person_name = person["name"].As<string>(), company_name = companyName});
            return 1;
        }));
    }
}

func addPersonsAsEmployees(driver neo4j.Driver, companyName string) (int, error) {
	session := driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
	defer session.Close()

	persons, err := neo4j.Collect(session.Run("MATCH (a:Person) RETURN a.name AS name", nil))
	if err != nil {
		return 0, err
	}

	employees := 0
	for _, person := range persons {
		_, err = session.WriteTransaction(func(tx neo4j.Transaction) (interface{}, error) {
			return tx.Run("MATCH (emp:Person {name: $person_name}) "+
				"MERGE (com:Company {name: $company_name}) "+
				"MERGE (emp)-[:WORKS_FOR]->(com)", map[string]interface{}{"person_name": person.Values[0], "company_name": companyName})
		})
		if err != nil {
			return 0, err
		}

		employees++
	}

	return employees, nil
}

public int addEmployees( final String companyName )
{
    try ( Session session = driver.session() )
    {
        int employees = 0;
        List<Record> persons = session.readTransaction( new TransactionWork<List<Record>>()
        {
            @Override
            public List<Record> execute( Transaction tx )
            {
                return matchPersonNodes( tx );
            }
        } );
        for ( final Record person : persons )
        {
            employees += session.writeTransaction( new TransactionWork<Integer>()
            {
                @Override
                public Integer execute( Transaction tx )
                {
                    tx.run( "MATCH (emp:Person {name: $person_name}) " +
                            "MERGE (com:Company {name: $company_name}) " +
                            "MERGE (emp)-[:WORKS_FOR]->(com)",
                            parameters( "person_name", person.get( "name" ).asString(), "company_name",
                                    companyName ) );
                    return 1;
                }
            } );
        }
        return employees;
    }
}

private static List<Record> matchPersonNodes( Transaction tx )
{
    return tx.run( "MATCH (a:Person) RETURN a.name AS name" ).list();
}

def add_employee_to_company(tx, person, company_name):
    tx.run("MATCH (emp:Person {name: $person_name}) "
           "MERGE (com:Company {name: $company_name}) "
           "MERGE (emp)-[:WORKS_FOR]->(com)",
           person_name=person["name"], company_name=company_name)
    return 1

def match_person_nodes(tx):
    return list(tx.run("MATCH (a:Person) RETURN a.name AS name"))

def add_employees(company_name):
    employees = 0
    with driver.session() as session:
        persons = session.read_transaction(match_person_nodes)

        for person in persons:
            employees += session.write_transaction(add_employee_to_company, person, company_name)

    return employees

异步会话

可用版本

.NET 驱动程序
Java 驱动程序
JavaScript 驱动程序

异步会话提供了一个 API，其中函数调用通常返回可用的对象，例如期货。这允许客户端应用程序在异步框架内工作并利用协作式多任务处理。

生命周期

会话生命周期从会话构造开始。然后，会话一直存在，直到它被关闭，这通常设置为在其包含的查询结果被使用后发生。

会话可以通过多种不同的方式进行配置。这通过在会话构造函数中提供配置来实现。有关更多详细信息，请参阅会话配置。

事务函数

事务函数**是包含事务性工作单元的推荐形式**。这种形式的事务需要最少的样板代码，并允许数据库查询和应用程序逻辑清晰分离。事务函数也很受欢迎，因为它们封装了重试逻辑，并且在将服务器的单个实例交换为集群时提供了最大程度的灵活性。

函数可以作为读取或写入操作调用。此选择会将事务路由到集群环境中的适当服务器。如果您在单实例环境中，则此路由没有影响，但如果您以后选择采用集群环境，它确实为您提供了灵活性。

在编写事务函数之前，务必确保事务函数执行的任何副作用都设计为幂等的。这是因为如果初始运行失败，则可能会多次执行函数。

当事务失败时，将调用驱动程序重试逻辑。对于某些故障情况，可以立即对另一台服务器重试事务。

这些情况包括连接问题、服务器角色更改（例如领导者选举）和瞬态错误。在创建会话时可以配置重试逻辑。

示例 6. 异步事务函数

事务函数

public async Task<List<string>> PrintAllProducts()
{
    List<string> result = null;
    var session = Driver.AsyncSession();

    try
    {
        // Wrap whole operation into an managed transaction and
        // get the results back.
        result = await session.ReadTransactionAsync(async tx =>
        {
            var products = new List<string>();

            // Send cypher query to the database
            var reader = await tx.RunAsync(
                "MATCH (p:Product) WHERE p.id = $id RETURN p.title", // Cypher query
                new {id = 0} // Parameters in the query, if any
            );

            // Loop through the records asynchronously
            while (await reader.FetchAsync())
            {
                // Each current read in buffer can be reached via Current
                products.Add(reader.Current[0].ToString());
            }

            return products;
        });
    }
    finally
    {
        // asynchronously close session
        await session.CloseAsync();
    }

    return result;
}

事务函数

public CompletionStage<ResultSummary> printAllProducts()
{
    String query = "MATCH (p:Product) WHERE p.id = $id RETURN p.title";
    Map<String,Object> parameters = Collections.singletonMap( "id", 0 );

    AsyncSession session = driver.asyncSession();

    return session.readTransactionAsync( tx ->
            tx.runAsync( query, parameters )
                    .thenCompose( cursor -> cursor.forEachAsync( record ->
                            // asynchronously print every record
                            System.out.println( record.get( 0 ).asString() ) ) )
    );
}

事务函数

const session = driver.session()
const titles = []
try {
  const result = await session.readTransaction(tx =>
    tx.run('MATCH (p:Product) WHERE p.id = $id RETURN p.title', { id: 0 })
  )

  const records = result.records
  for (let i = 0; i < records.length; i++) {
    const title = records[i].get(0)
    titles.push(title)
  }
} finally {
  await session.close()
}

自动提交事务

自动提交事务是一种基本但有限的事务形式。此类事务仅包含一个 Cypher 查询，并且在失败时不会自动重放。因此，任何错误情况都必须由客户端应用程序本身处理。

自动提交事务旨在用于简单的用例，例如学习 Cypher 或编写一次性脚本时。

不建议在生产环境中使用自动提交事务。

与其他类型的 Cypher 查询不同，PERIODIC COMMIT 查询不参与因果链。

因此，从驱动程序执行它们的唯一方法是使用自动提交事务。

请参阅Cypher 手册 → PERIODIC COMMIT 查询提示。

示例 7. 异步自动提交事务

自动提交事务

public async Task<List<string>> ReadProductTitles()
{
    var records = new List<string>();
    var session = Driver.AsyncSession();

    try
    {
        // Send cypher query to the database.
        // The existing IResult interface implements IEnumerable
        // and does not play well with asynchronous use cases. The replacement
        // IResultCursor interface is returned from the RunAsync
        // family of methods instead and provides async capable methods. 
        var reader = await session.RunAsync(
            "MATCH (p:Product) WHERE p.id = $id RETURN p.title", // Cypher query
            new {id = 0} // Parameters in the query, if any
        );

        // Loop through the records asynchronously
        while (await reader.FetchAsync())
        {
            // Each current read in buffer can be reached via Current
            records.Add(reader.Current[0].ToString());
        }
    }
    finally
    {
        // asynchronously close session
        await session.CloseAsync();
    }

    return records;
}

自动提交事务

public CompletionStage<List<String>> readProductTitles()
{
    String query = "MATCH (p:Product) WHERE p.id = $id RETURN p.title";
    Map<String,Object> parameters = Collections.singletonMap( "id", 0 );

    AsyncSession session = driver.asyncSession();

    return session.runAsync( query, parameters )
            .thenCompose( cursor -> cursor.listAsync( record -> record.get( 0 ).asString() ) )
            .exceptionally( error ->
            {
                // query execution failed, print error and fallback to empty list of titles
                error.printStackTrace();
                return Collections.emptyList();
            } )
            .thenCompose( titles -> session.closeAsync().thenApply( ignore -> titles ) );
}

自动提交事务

async function readProductTitles () {
  const session = driver.session()
  try {
    const result = await session.run(
      'MATCH (p:Product) WHERE p.id = $id RETURN p.title',
      {
        id: 0
      }
    )
    const records = result.records
    const titles = []
    for (let i = 0; i < records.length; i++) {
      const title = records[i].get(0)
      titles.push(title)
    }
    return titles
  } finally {
    await session.close()
  }
}

使用结果

异步会话 API 提供了特定于语言的方法来帮助与异步应用程序和框架集成。

示例 8. 异步使用结果

使用结果

public async Task<List<string>> GetPeopleAsync()
{
    var session = Driver.AsyncSession();
    try
    {
        return await session.ReadTransactionAsync(async tx =>
        {
            var result = await tx.RunAsync("MATCH (a:Person) RETURN a.name ORDER BY a.name");

            return await result.ToListAsync(r => r[0].As<string>());
        });
    }
    finally
    {
        await session.CloseAsync();
    }
}

使用结果

public CompletionStage<List<String>> getPeople()
{

    String query = "MATCH (a:Person) RETURN a.name ORDER BY a.name";
    AsyncSession session = driver.asyncSession();
    return session.readTransactionAsync( tx ->
            tx.runAsync( query )
                    .thenCompose( cursor -> cursor.listAsync( record ->
                            record.get( 0 ).asString() ) )
    );
}

使用结果

const session = driver.session()
const result = session.run('MATCH (a:Person) RETURN a.name ORDER BY a.name')
const collectedNames = []

result.subscribe({
  onNext: record => {
    const name = record.get(0)
    collectedNames.push(name)
  },
  onCompleted: () => {
    session.close().then(() => {
      console.log('Names: ' + collectedNames.join(', '))
    })
  },
  onError: error => {
    console.log(error)
  }
})

反应式会话

可用版本

.NET 驱动程序
Java 驱动程序
JavaScript 驱动程序

从**Neo4j 4.0** 开始，支持查询的反应式处理。这可以通过反应式会话来实现。反应式会话允许动态管理在驱动程序和服务器之间交换的数据。

响应式编程的典型特征是，消费者控制从查询中获取记录的速度，而驱动程序则管理从服务器请求记录的速度。流控制在整个 Neo4j 堆栈中都得到支持，这意味着查询引擎会对流控制信号做出正确的响应。这使得资源处理更加高效，并确保接收方不会被迫缓冲任意数量的数据。

有关响应式流的更多信息，请参阅以下内容：

响应式会话通常用于已经面向响应式风格的客户端应用程序；预计已存在响应式依赖项或框架。

有关推荐依赖项的更多信息，请参阅入门。

生命周期

会话生命周期从会话构造开始。然后，会话一直存在，直到它被关闭，这通常设置为在其包含的查询结果被使用后发生。

事务函数

这种形式的事务需要最少的样板代码，并允许数据库查询和应用程序逻辑之间进行清晰的分离。事务函数也很理想，因为它们封装了重试逻辑，并在将服务器的单个实例替换为集群时提供了最大的灵活性。

在编写事务函数之前，务必确保事务函数执行的任何副作用都设计为幂等的。这是因为如果初始运行失败，则可能会多次执行函数。

当事务失败时，将调用驱动程序重试逻辑。对于某些失败情况，可以立即对不同的服务器重试事务。这些情况包括连接问题、服务器角色更改（例如领导者选举）和瞬态错误。在创建会话时可以配置重试逻辑。

示例 9. 响应式事务函数

响应式事务函数

public IObservable<string> PrintAllProducts()
{
    var session = Driver.RxSession();

    return session.ReadTransaction(tx =>
    {
        return tx.Run(
                "MATCH (p:Product) WHERE p.id = $id RETURN p.title", // Cypher query
                new {id = 0} // Parameters in the query, if any
            )
            .Records()
            .Select(record => record[0].ToString());
    }).OnErrorResumeNext(session.Close<string>());
}

响应式事务函数

public Flux<ResultSummary> printAllProducts()
{
    String query = "MATCH (p:Product) WHERE p.id = $id RETURN p.title";
    Map<String,Object> parameters = Collections.singletonMap( "id", 0 );

    return Flux.usingWhen( Mono.fromSupplier( driver::rxSession ),
            session -> session.readTransaction( tx -> {
                RxResult result = tx.run( query, parameters );
                return Flux.from( result.records() )
                        .doOnNext( record -> System.out.println( record.get( 0 ).asString() ) ).then( Mono.from( result.consume() ) );
            }
         ), RxSession::close );
}

响应式事务函数

const session = driver.rxSession()
const result = session.readTransaction(tx =>
  tx
    .run('MATCH (p:Product) WHERE p.id = $id RETURN p.title', { id: 0 })
    .records()
    .pipe(
      map(r => r.get(0)),
      materialize(),
      toArray()
    )
)

会话可以通过多种不同的方式进行配置。这通过在会话构造函数中提供配置来实现。有关更多详细信息，请参阅会话配置。

自动提交事务

自动提交事务是一种基本但有限的事务形式。这种事务仅包含一个 Cypher 查询，并且不会在失败时自动重放。因此，任何错误场景都需要由客户端应用程序本身处理。

自动提交事务旨在用于简单的用例，例如学习 **Cypher** 或编写一次性脚本。

不建议在生产环境中使用自动提交事务。

执行 PERIODIC COMMIT Cypher 查询的唯一方法是自动提交事务。

与其他类型的 Cypher 查询不同，PERIODIC COMMIT 查询不参与因果链。

请参阅Cypher 手册 → PERIODIC COMMIT 查询提示。

示例 10. 响应式自动提交事务

自动提交事务

public IObservable<string> ReadProductTitles()
{
    var session = Driver.RxSession();

    return session.Run(
            "MATCH (p:Product) WHERE p.id = $id RETURN p.title", // Cypher query
            new {id = 0} // Parameters in the query, if any
        )
        .Records()
        .Select(record => record[0].ToString())
        .OnErrorResumeNext(session.Close<string>());
}

自动提交事务

public Flux<String> readProductTitles()
{
    String query = "MATCH (p:Product) WHERE p.id = $id RETURN p.title";
    Map<String,Object> parameters = Collections.singletonMap( "id", 0 );

    return Flux.usingWhen( Mono.fromSupplier( driver::rxSession ),
            session -> Flux.from( session.run( query, parameters ).records() ).map( record -> record.get( 0 ).asString() ),
            RxSession::close );
}

自动提交事务

function readProductTitles () {
  const session = driver.rxSession()
  return session
    .run('MATCH (p:Product) WHERE p.id = $id RETURN p.title', {
      id: 0
    })
    .records()
    .pipe(
      map(r => r.get(0)),
      materialize(),
      toArray()
    )
}

使用结果

要在响应式会话中使用查询中的数据，需要一个订阅者来处理发布者返回的结果。

每个事务对应一个数据流，该数据流提供来自服务器的数据。当从该流中提取记录时，结果处理开始。只有一个订阅者可以从给定流中提取数据。

示例 11. 响应式使用结果

使用结果

public IObservable<string> GetPeople()
{
    var session = Driver.RxSession();
    return session.ReadTransaction(tx =>
    {
        return tx.Run("MATCH (a:Person) RETURN a.name ORDER BY a.name")
            .Records()
            .Select(record => record[0].As<string>());
    }).OnErrorResumeNext(session.Close<string>());
}

使用结果

public Flux<String> getPeople()
{
    String query = "MATCH (a:Person) RETURN a.name ORDER BY a.name";

    return Flux.usingWhen( Mono.fromSupplier( driver::rxSession ),
            session -> session.readTransaction( tx -> {
                        RxResult result = tx.run( query );
                        return Flux.from( result.records() )
                                .map( record -> record.get( 0 ).asString() );
                    }
            ), RxSession::close );
}

使用结果

const session = driver.rxSession()
const result = session
  .run('MATCH (a:Person) RETURN a.name ORDER BY a.name')
  .records()
  .pipe(
    map(r => r.get(0)),
    materialize(),
    toArray()
  )

取消

根据响应式流规范，可以在过程中取消响应式数据流。这会过早地提交或回滚事务，并阻止查询引擎生成更多记录。

会话配置

书签: 确保会话内事务之间因果一致性的机制。书签在单个会话内的事务之间隐式传递，以满足因果一致性要求。在某些情况下，您可能希望在一个会话中使用书签在另一个新的会话中使用。

默认值：无（会话最初将没有书签创建）
默认访问模式: 当不使用事务函数时，访问模式设置的回退。通常，通过调用相应的事务函数方法为每个事务设置访问模式。在其他情况下，此设置将被继承。请注意，事务函数将忽略/覆盖此设置。

默认值：写
数据库: 会话将与其交互的数据库。当您使用非默认数据库（即 system 数据库或 Neo4j 4.0 企业版中的其他数据库）时，您可以显式配置驱动程序正在对其执行事务的数据库。有关数据库的更多信息，请参阅操作手册 → 默认数据库。

默认值：服务器上配置的默认数据库。
获取大小: 从服务器的每个批次中获取的记录数。Neo4j 4.0 引入了按批次提取记录的功能，允许客户端应用程序控制数据填充并对服务器应用反压。此 FetchSize 应用于简单会话和异步会话，而响应式会话可以使用订阅的请求方法直接控制。

默认值：1000 条记录