Kafka

APOC Kafka 过程

要启用 Kafka 依赖项，我们需要设置 APOC 配置 apoc.kafka.enabled=true

任何以 apoc.kafka. 开头的配置选项都控制过程本身的行为。

Kafka 依赖项包含在 apoc-kafka-dependencies-5.26.1-all.jar 文件中，可以从版本发布页面下载。下载该文件后，应将其放置在 plugins 目录中，并重启 Neo4j 服务器。

任何以 apoc.kafka. 开头的配置选项都将传递给底层的 Kafka 驱动程序。Neo4j Kafka 过程使用官方的 Confluent Kafka 生产者和消费者 Java 客户端。对这些连接器有效的配置设置也适用于 APOC Kafka。

例如，在下方链接的 Kafka 文档中，名为 batch.size 的配置设置在 APOC Kafka 中应表示为 apoc.kafka.batch.size。

以下是您可能希望使用的常见配置设置。.最常见的必需配置设置

设置名称	描述	默认值
apoc.kafka.max.poll.records	从 Kafka 每批拉取的最大记录数。增加此数字意味着 Neo4j 内存中的事务更大，并可能提高吞吐量。	500
apoc.kafka.buffer.memory	生产者可用于缓冲等待记录的总内存字节数。使用此项调整过程可能需要多少内存来保存尚未传递到 Neo4j 的消息。	33554432
apoc.kafka.batch.size	(仅生产者) 当向同一分区发送多条记录时，生产者将尝试将记录批量处理到更少的请求中。这有助于提高客户端和服务器的性能。此配置控制默认的批量大小（以字节为单位）。	16384
apoc.kafka.max.partition.fetch.bytes	(仅消费者) 服务器将返回的每个分区的最大数据量。消费者按批次获取记录。如果获取的第一个非空分区的第一个记录批次大于此限制，该批次仍将返回，以确保消费者能够继续进行。	1048576
apoc.kafka.group.id	唯一标识此消费者所属的消费者组的字符串。	不适用

设置名称

描述

默认值

apoc.kafka.max.poll.records

从 Kafka 每批拉取的最大记录数。增加此数字意味着 Neo4j 内存中的事务更大，并可能提高吞吐量。

500

apoc.kafka.buffer.memory

生产者可用于缓冲等待记录的总内存字节数。使用此项调整过程可能需要多少内存来保存尚未传递到 Neo4j 的消息。

33554432

apoc.kafka.batch.size

(仅生产者) 当向同一分区发送多条记录时，生产者将尝试将记录批量处理到更少的请求中。这有助于提高客户端和服务器的性能。此配置控制默认的批量大小（以字节为单位）。

16384

apoc.kafka.max.partition.fetch.bytes

(仅消费者) 服务器将返回的每个分区的最大数据量。消费者按批次获取记录。如果获取的第一个非空分区的第一个记录批次大于此限制，该批次仍将返回，以确保消费者能够继续进行。

1048576

apoc.kafka.group.id

唯一标识此消费者所属的消费者组的字符串。

不适用

如果您在本地或独立机器上运行，请配置 apoc.conf 指向该服务器

neo4j.conf

apoc.kafka.bootstrap.servers=localhost:9092

如果您使用 Confluent Cloud（托管 Kafka），您可以按照Confluent Cloud 部分的描述连接到 Kafka

插件安装和配置完成后，重启数据库将使其激活。如果您已配置 Neo4j 从 kafka 消费，它将立即开始处理消息。