如何从Neo4j 4.3企业版直接向Kafka发布自定义消息

针对你遇到的Neo4j Streams插件版本兼容问题，这里有几个可靠的方案可以实现从Neo4j 4.3直接发布自定义数据到Kafka：

方案1：通过APOC扩展调用Kafka Java客户端 #

APOC（Awesome Procedures on Cypher）支持调用Java代码，而Kafka提供了成熟的Java客户端，我们可以利用这一点在Cypher中直接发送Kafka消息。

步骤： #

1. 准备依赖：

   • 下载与你的Kafka版本匹配的kafka-clients.jar（比如Kafka 2.8.x对应kafka-clients-2.8.2.jar），将其放到Neo4j的plugins目录下。
   • 在neo4j.conf中添加配置，允许APOC调用Java代码并加载Kafka客户端：

     apoc.jvm.additional=-cp plugins/kafka-clients-2.8.2.jar
     apoc.java.enabled=true
     

   • 重启Neo4j服务。

2. 编写Cypher查询：
   利用apoc.java.run来初始化Kafka Producer并发送消息，结合apoc.periodic.iterate处理批量数据：

   CALL apoc.periodic.iterate(
     'MATCH (u:User {status: "VERIFIED"}) RETURN u',
     'CALL apoc.java.run(
       "org.apache.kafka.clients.producer.KafkaProducer",
       [
         {bootstrap.servers: "your-kafka-broker:9092"},
         {acks: "all"}
       ],
       "send",
       [
         org.apache.kafka.clients.producer.ProducerRecord,
         "test-topic",
         u.name,
         apoc.convert.toJson({name: u.name})
       ]
     )',
     {parallel: false}
   )
   

   注意：如果需要复用Producer（避免每次调用创建新实例），可以通过APOC的apoc.singleton来维护一个全局的Producer实例。

方案2：自定义Neo4j存储过程 #

如果APOC的Java调用不够灵活，你可以自己编写一个Java存储过程，集成Kafka Producer逻辑，部署到Neo4j中使用。

步骤： #

1. 创建Maven项目：
   添加Neo4j过程API和Kafka客户端依赖：

   <dependencies>
       <dependency>
           <groupId>org.neo4j</groupId>
           <artifactId>neo4j-procedure-api</artifactId>
           <version>4.3.22</version>
           <scope>provided</scope>
       </dependency>
       <dependency>
           <groupId>org.apache.kafka</groupId>
           <artifactId>kafka-clients</artifactId>
           <version>2.8.2</version>
       </dependency>
   </dependencies>
   

2. 编写存储过程：

   package com.example.neo4j.kafka;
   
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   import org.neo4j.procedure.*;
   
   import java.util.Properties;
   
   public class KafkaPublishProcedure {
       private static KafkaProducer<String, String> producer;
   
       @SuppressWarnings("unused")
       @Procedure(name = "com.example.kafka.publish", mode = Mode.WRITE)
       @Description("CALL com.example.kafka.publish(topic, key, message) - Publishes a message to Kafka")
       public void publish(
               @Name("topic") String topic,
               @Name("key") String key,
               @Name("message") String message) {
           if (producer == null) {
               Properties props = new Properties();
               props.put("bootstrap.servers", "your-kafka-broker:9092");
               props.put("acks", "all");
               props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
               props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
               producer = new KafkaProducer<>(props);
           }
           producer.send(new ProducerRecord<>(topic, key, message));
       }
   }
   

3. 部署与使用：

   • 打包项目成jar文件，放到Neo4j的plugins目录。
   • 在neo4j.conf中添加：

     dbms.security.procedures.unrestricted=com.example.kafka.*
     

   • 重启Neo4j后，就可以在Cypher中调用：

     CALL apoc.periodic.iterate(
       'MATCH (u:User {status: "VERIFIED"}) RETURN u',
       'CALL com.example.kafka.publish("test-topic", u.name, apoc.convert.toJson({name: u.name}))',
       {parallel: false}
     )
     

方案3：利用Neo4j CDC + Kafka Connect（适合基于变更的场景） #

如果你的自定义数据是基于节点/关系的变更触发的，可以使用Neo4j 4.3内置的CDC功能，结合Debezium的Neo4j源连接器来捕获变更并发送到Kafka。不过这个方案更适合实时捕获数据变更，而非主动查询并发布自定义数据，但如果你的需求可以通过标记节点状态（比如新增一个to_publish属性）来触发，也可以适用。

核心步骤： #

1. 在neo4j.conf中启用CDC：

   dbms.change_data_capture.enabled=true
   

2. 配置Debezium Neo4j源连接器，捕获指定标签的节点变更，过滤出需要发布的记录，发送到目标Kafka主题。

注意事项 #

• 无论使用哪种方案，都要确保Kafka集群与Neo4j网络连通，并且Producer的配置（如bootstrap.servers）正确。
• 批量处理时，要注意控制并发数，避免给Kafka或Neo4j造成过大压力。
• 对于自定义存储过程，建议将Kafka配置参数放到neo4j.conf中，通过Config对象读取，避免硬编码。

内容的提问来源于stack exchange，提问作者flaviuratiu