什么是 ‘Asynchronous Edge Propagation’：在高并发图中，实现非阻塞式的状态传播与副作用处理

各位同仁，下午好！

今天，我们将深入探讨一个在构建高并发、大规模图系统时至关重要的技术范式：Asynchronous Edge Propagation (AEP)，即非阻塞式的状态传播与副作用处理。在现代互联网应用中，图结构无处不在，从社交网络的好友关系到推荐系统的物品关联，从金融交易的依赖链到微服务架构的服务调用图，图的动态性与复杂性对系统的伸缩性和响应性提出了严峻挑战。当图中的某个节点或边发生变化时，这种变化往往会沿着图的结构“传播”开来，触发一系列链式反应，并可能导致多种外部副作用。如何高效、可靠、非阻塞地处理这些传播和副作用，正是AEP的核心使命。

图结构与传播的本质

在深入AEP之前，我们首先要对图以及“传播”这个概念有一个清晰的认识。

图的定义：
一个图 $G = (V, E)$ 由一组顶点（或节点）$V$ 和一组边 $E$ 组成。

• 节点 (Nodes/Vertices): 代表实体，例如用户、产品、文章、服务等。节点可以拥有属性（properties），如用户ID、姓名、创建时间、状态等。
• 边 (Edges/Relationships): 代表实体之间的关系，例如“关注”、“购买”、“依赖于”、“属于”。边也可以拥有属性，如关注时间、购买数量、关系强度等。

传播的含义：
图中的“传播”是指当某个节点或边的状态发生改变时，这种变化会根据预定义的规则，通过其相邻的边或节点，影响到图中的其他部分。这种影响可以是：

1. 状态更新： 某个节点的状态变化导致其邻居节点也需要更新状态。
2. 派生计算： 某个属性的改变触发了对相关节点或边的派生属性的重新计算。
3. 副作用触发： 图内部的变化需要通知外部系统，例如发送通知、更新缓存、触发外部服务调用等。

例子：社交网络的用户关注

• 用户A关注了用户B。这是一个边创建事件。
• 用户B的“粉丝数”节点属性需要更新。
• 用户A的“关注数”节点属性需要更新。
• 可能需要向用户B发送“A关注了你”的通知（副作用）。
• 可能需要更新用户A的个性化推荐列表（副作用）。
• 可能需要将用户B的最新动态推送到用户A的关注流中（副作用）。

可以看到，一个简单的操作，就会导致图内部的多个状态更新和多个外部副作用。在低并发场景下，我们或许可以采取同步处理的方式。但当每秒有数千甚至数万个这样的操作发生时，同步处理将迅速成为性能瓶颈，导致用户体验下降，甚至系统崩溃。

传统同步传播的困境

在没有AEP的情况下，一个直观的做法是：当一个图操作发生时，立即在同一个事务或同一个线程中完成所有相关的状态传播和副作用处理。

// 传统同步传播示例
public class SynchronousGraphService {

    private GraphDatabase graphDb; // 假设这是一个图数据库接口
    private NotificationService notificationService;
    private CacheService cacheService;

    public SynchronousGraphService(GraphDatabase graphDb, NotificationService notificationService, CacheService cacheService) {
        this.graphDb = graphDb;
        this.notificationService = notificationService;
        this.cacheService = cacheService;
    }

    public void userFollows(String followerId, String followeeId) {
        // 1. 核心图数据更新
        graphDb.createEdge(followerId, followeeId, "FOLLOWS");

        // 2. 状态传播：更新粉丝数/关注数
        graphDb.incrementNodeProperty(followeeId, "followersCount");
        graphDb.incrementNodeProperty(followerId, "followingCount");

        // 3. 副作用处理：发送通知
        notificationService.sendNotification(followeeId, followerId + " started following you.");

        // 4. 副作用处理：更新缓存
        cacheService.invalidateUserFeedCache(followerId);
        cacheService.invalidateUserProfileCache(followeeId);

        // ... 更多可能的同步操作
    }
}

这种同步方式在高并发场景下会面临一系列严重问题：

1. 性能瓶颈： 每一个图操作都需要等待所有传播和副作用处理完成后才能返回。如果任何一个下游操作（如发送通知、复杂计算）耗时较长，整个主操作的响应时间就会被拉长。
2. 资源争抢与死锁： 在更新多个节点或边时，需要对相关资源加锁以保证数据一致性。在高并发下，频繁的锁竞争会导致严重的性能下降，甚至产生死锁。
3. 事务复杂性： 如果传播和副作用处理涉及多个外部系统（如数据库、消息队列、缓存），将其全部纳入一个分布式事务中将极其复杂，且容易失败。
4. 系统耦合： 核心图服务与各种副作用处理服务紧密耦合。任何一个下游服务的故障或性能问题都可能影响到核心服务。
5. 可伸缩性差： 核心服务和传播/副作用服务难以独立伸缩。当某个环节出现瓶颈时，整个系统都受影响。
6. 弹性不足： 如果某个副作用处理失败（例如通知服务暂时不可用），整个用户操作可能会回滚或失败，影响用户体验。

这些问题在高并发图系统中尤为突出，因为图的连接性意味着一个小的变化可能引发“雪崩式”的传播效应。因此，我们需要一种非阻塞的、解耦的机制来处理这些传播和副作用，这就是Asynchronous Edge Propagation的用武之地。

Asynchronous Edge Propagation (AEP) 的核心理念

AEP的核心思想是解耦。它将图的核心状态变更与状态传播及副作用处理这两个阶段分离。当图中的状态发生改变时，核心服务只负责记录这个变化，并发出一个事件（或消息），然后立即返回。真正的传播和副作用处理则由独立的、异步的消费者服务在后台完成。

AEP的关键特征：

1. 事件驱动 (Event-Driven): 图中的每一个有意义的变化都被封装成一个事件。
2. 消息队列/事件流 (Message Queue/Event Stream): 事件通过可靠的消息中间件进行传递。
3. 非阻塞 (Non-blocking): 主操作线程不等待传播和副作用处理的结果。
4. 最终一致性 (Eventual Consistency): 传播后的状态不保证立即一致，但最终会达到一致。
5. 独立伸缩 (Independent Scalability): 各个组件可以独立部署和伸缩。
6. 弹性与容错 (Resilience & Fault Tolerance): 利用消息队列的重试机制和死信队列，提高系统健壮性。

AEP的架构组件

一个典型的AEP架构包含以下核心组件：

组件名称	职责	常用技术栈
图数据存储 (Graph Data Store)	存储图的节点和边数据。负责核心图操作的原子性和持久化。在AEP中，它是事件的生产者。	Neo4j, JanusGraph, ArangoDB, Amazon Neptune, Apache TinkerPop, 自定义KV存储上的图层
事件发布器 (Event Publisher)	当核心图数据发生变化时，将变化信息封装成事件并发布到消息队列。通常是图服务内部的一部分。	集成消息队列客户端的业务逻辑
消息中间件 (Message Broker/Queue)	负责可靠地接收、存储和分发事件。提供持久化、顺序保证（可选）、发布/订阅、消费者组等功能。是解耦的核心。	Apache Kafka, RabbitMQ, Amazon SQS/SNS, Azure Service Bus, Google Cloud Pub/Sub
事件消费者/工作器 (Event Consumers/Workers)	订阅消息队列中的事件，解析事件内容，并根据业务逻辑执行状态传播（更新图中的其他节点/边）和副作用处理（调用外部服务、发送通知、更新缓存等）。消费者可以有多种类型，各自负责不同的传播或副作用。	Spring Boot应用, Go服务, Python脚本, Lambda函数等，集成消息队列客户端
副作用处理器 (Side Effect Handlers)	通常是事件消费者的一部分，或者由事件消费者触发的下游服务。负责与外部系统交互，完成非图状态的更新。	独立的微服务、邮件/短信服务、缓存服务、日志服务、AI推荐服务等
状态追踪/协调器 (State Tracker/Coordinator) (可选)	对于复杂的传播链，可能需要一个机制来追踪整个传播过程是否完成，或者协调多个消费者之间的协作。通常在最终一致性要求较高的场景下使用。	分布式事务框架、Saga模式协调器、基于ZooKeeper/etcd的分布式锁或状态机、独立的跟踪服务

AEP的实现细节与代码示例

为了更好地理解AEP，我们通过一个具体的Java示例来演示其实现。假设我们正在构建一个社交网络，当用户A关注用户B时，需要：

1. 在图数据库中记录关注关系。
2. 更新用户A的关注数 (followingCount)。
3. 更新用户B的粉丝数 (followersCount)。
4. 向用户B发送关注通知。
5. 更新用户A的个性化推荐缓存。

我们将使用Spring Boot作为应用程序框架，Apache Kafka作为消息中间件。

1. 定义事件模型

首先，定义一个通用的图变更事件基类，以及具体的关注事件。

// graph-common/src/main/java/com/example/graph/event/GraphChangeEvent.java
package com.example.graph.event;

import java.io.Serializable;
import java.time.Instant;

public abstract class GraphChangeEvent implements Serializable {
    private static final long serialVersionUID = 1L;

    private String eventId;
    private Instant timestamp;
    private String eventType; // 例如 "NODE_UPDATED", "EDGE_CREATED"

    public GraphChangeEvent() {
        this.eventId = java.util.UUID.randomUUID().toString();
        this.timestamp = Instant.now();
        this.eventType = this.getClass().getSimpleName();
    }

    // Getters
    public String getEventId() { return eventId; }
    public Instant getTimestamp() { return timestamp; }
    public String getEventType() { return eventType; }

    @Override
    public String toString() {
        return "GraphChangeEvent{" +
               "eventId='" + eventId + ''' +
               ", timestamp=" + timestamp +
               ", eventType='" + eventType + ''' +
               '}';
    }
}

// graph-common/src/main/java/com/example/graph/event/UserFollowsEvent.java
package com.example.graph.event;

public class UserFollowsEvent extends GraphChangeEvent {
    private static final long serialVersionUID = 1L;

    private String followerId;
    private String followeeId;

    public UserFollowsEvent() {
        // For deserialization
    }

    public UserFollowsEvent(String followerId, String followeeId) {
        super();
        this.followerId = followerId;
        this.followeeId = followeeId;
    }

    // Getters
    public String getFollowerId() { return followerId; }
    public String getFolloweeId() { return followeeId; }

    @Override
    public String toString() {
        return "UserFollowsEvent{" +
               "eventId='" + getEventId() + ''' +
               ", timestamp=" + getTimestamp() +
               ", followerId='" + followerId + ''' +
               ", followeeId='" + followeeId + ''' +
               '}';
    }
}

2. 事件发布器：核心图服务

核心图服务负责处理用户请求，更新图数据库，并发布事件。它不直接处理任何传播或副作用。

// graph-core-service/src/main/java/com/example/graph/core/service/GraphCoreService.java
package com.example.graph.core.service;

import com.example.graph.event.UserFollowsEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class GraphCoreService {

    private static final Logger log = LoggerFactory.getLogger(GraphCoreService.class);
    private final GraphDatabase graphDb; // 模拟图数据库接口
    private final KafkaTemplate<String, UserFollowsEvent> kafkaTemplate;
    private final String KAFKA_TOPIC = "graph-events";

    public GraphCoreService(GraphDatabase graphDb, KafkaTemplate<String, UserFollowsEvent> kafkaTemplate) {
        this.graphDb = graphDb;
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 用户关注操作的核心逻辑。
     * 仅负责更新图数据库的核心关系，并发布事件。
     * 不处理任何计数更新、通知、缓存等副作用。
     */
    public void userFollows(String followerId, String followeeId) {
        // 1. 在图数据库中创建关注边
        // 假设 graphDb.createEdge 是一个原子操作
        graphDb.createEdge(followerId, followeeId, "FOLLOWS");
        log.info("用户 {} 关注了用户 {}，核心关系已建立。", followerId, followeeId);

        // 2. 发布事件到Kafka
        UserFollowsEvent event = new UserFollowsEvent(followerId, followeeId);
        kafkaTemplate.send(KAFKA_TOPIC, event.getEventId(), event); // 使用 eventId 作为 key 保证同一事件在同一分区
        log.info("UserFollowsEvent {} 已发布到Kafka。", event.getEventId());
    }

    // 模拟图数据库接口
    public interface GraphDatabase {
        void createEdge(String sourceId, String targetId, String relationshipType);
        // ... 其他核心图操作
    }

    // 实际应用中，这里会注入一个真实的图数据库客户端，例如 Neo4j Driver
    @Service
    public static class MockGraphDatabase implements GraphDatabase {
        @Override
        public void createEdge(String sourceId, String targetId, String relationshipType) {
            log.debug("MockGraphDatabase: Creating edge {} -[{}]-> {}", sourceId, relationshipType, targetId);
            // 模拟数据库操作耗时
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

在GraphCoreService中，userFollows方法首先更新核心图数据，然后立即发布一个UserFollowsEvent到Kafka。整个过程是快速且非阻塞的。

3. 事件消费者：处理计数更新

这个消费者负责订阅graph-events主题，处理UserFollowsEvent，并更新用户的关注数和粉丝数。

// graph-propagation-service/src/main/java/com/example/graph/propagation/service/CountUpdateConsumer.java
package com.example.graph.propagation.service;

import com.example.graph.event.UserFollowsEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class CountUpdateConsumer {

    private static final Logger log = LoggerFactory.getLogger(CountUpdateConsumer.class);
    private final GraphPropertyService graphPropertyService; // 模拟更新节点属性的服务

    public CountUpdateConsumer(GraphPropertyService graphPropertyService) {
        this.graphPropertyService = graphPropertyService;
    }

    @KafkaListener(topics = "graph-events", groupId = "count-update-group", containerFactory = "kafkaListenerContainerFactory")
    public void listen(UserFollowsEvent event) {
        log.info("收到 UserFollowsEvent ({}): {} 关注了 {}", event.getEventId(), event.getFollowerId(), event.getFolloweeId());

        try {
            // 更新关注者的关注数
            graphPropertyService.incrementNodeProperty(event.getFollowerId(), "followingCount");
            log.info("已更新用户 {} 的 followingCount。", event.getFollowerId());

            // 更新被关注者的粉丝数
            graphPropertyService.incrementNodeProperty(event.getFolloweeId(), "followersCount");
            log.info("已更新用户 {} 的 followersCount。", event.getFolloweeId());

        } catch (Exception e) {
            log.error("处理 UserFollowsEvent {} 时发生错误：{}", event.getEventId(), e.getMessage(), e);
            // 实际生产中，这里可能需要将事件发送到死信队列 (DLQ) 进行后续处理
            // 或者实现指数退避重试机制
        }
    }

    // 模拟图属性服务
    public interface GraphPropertyService {
        void incrementNodeProperty(String nodeId, String propertyName);
    }

    @Service
    public static class MockGraphPropertyService implements GraphPropertyService {
        @Override
        public void incrementNodeProperty(String nodeId, String propertyName) {
            log.debug("MockGraphPropertyService: Incrementing {} for node {}", propertyName, nodeId);
            // 模拟数据库属性更新操作耗时
            try {
                Thread.sleep(30);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

这个消费者通过@KafkaListener注解监听Kafka主题。它独立于核心服务运行，即使核心服务正在处理大量请求，它也能在后台慢慢消费事件。

4. 事件消费者：处理通知与缓存更新（副作用处理）

这个消费者负责处理发送通知和更新缓存的副作用。它同样订阅graph-events主题。

// graph-sideeffect-service/src/main/java/com/example/graph/sideeffect/service/SideEffectConsumer.java
package com.example.graph.sideeffect.service;

import com.example.graph.event.UserFollowsEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class SideEffectConsumer {

    private static final Logger log = LoggerFactory.getLogger(SideEffectConsumer.class);
    private final NotificationService notificationService;
    private final CacheService cacheService;

    public SideEffectConsumer(NotificationService notificationService, CacheService cacheService) {
        this.notificationService = notificationService;
        this.cacheService = cacheService;
    }

    @KafkaListener(topics = "graph-events", groupId = "side-effect-group", containerFactory = "kafkaListenerContainerFactory")
    public void listen(UserFollowsEvent event) {
        log.info("收到 UserFollowsEvent ({}): 处理副作用 {} 关注了 {}", event.getEventId(), event.getFollowerId(), event.getFolloweeId());

        try {
            // 副作用：发送通知
            notificationService.sendNotification(event.getFolloweeId(), event.getFollowerId() + " 关注了你！");
            log.info("已向 {} 发送关注通知。", event.getFolloweeId());

            // 副作用：更新缓存
            cacheService.invalidateUserFeedCache(event.getFollowerId());
            cacheService.invalidateUserProfileCache(event.getFolloweeId());
            log.info("已刷新用户 {} 和 {} 的相关缓存。", event.getFollowerId(), event.getFolloweeId());

        } catch (Exception e) {
            log.error("处理 UserFollowsEvent {} 的副作用时发生错误：{}", event.getEventId(), e.getMessage(), e);
            // 同样，考虑死信队列和重试
        }
    }

    // 模拟通知服务
    public interface NotificationService {
        void sendNotification(String userId, String message);
    }

    @Service
    public static class MockNotificationService implements NotificationService {
        @Override
        public void sendNotification(String userId, String message) {
            log.debug("MockNotificationService: Sending notification to {}: {}", userId, message);
            // 模拟发送通知耗时
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 模拟缓存服务
    public interface CacheService {
        void invalidateUserFeedCache(String userId);
        void invalidateUserProfileCache(String userId);
    }

    @Service
    public static class MockCacheService implements CacheService {
        @Override
        public void invalidateUserFeedCache(String userId) {
            log.debug("MockCacheService: Invalidating feed cache for {}", userId);
            try {
                Thread.sleep(20);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        @Override
        public void invalidateUserProfileCache(String userId) {
            log.debug("MockCacheService: Invalidating profile cache for {}", userId);
            try {
                Thread.sleep(20);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

5. Kafka配置与序列化

为了让Spring Boot应用能够与Kafka交互，需要进行配置。事件对象需要被序列化和反序列化。通常我们会使用JSON或Avro。这里以JSON为例。

// graph-common/src/main/java/com/example/graph/config/KafkaConfig.java
package com.example.graph.config;

import com.example.graph.event.GraphChangeEvent;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
import static org.apache.kafka.clients.producer.ProducerConfig.*;

@Configuration
public class KafkaConfig {

    // Kafka服务器地址
    private final String bootstrapServers = "localhost:9092";

    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new JavaTimeModule()); // 支持Java 8日期时间类型
        return mapper;
    }

    // Producer Configuration
    @Bean
    public ProducerFactory<String, GraphChangeEvent> producerFactory(ObjectMapper objectMapper) {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 配置JsonSerializer使用我们自定义的ObjectMapper
        configProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true); // 在消息头中添加类型信息，方便反序列化
        configProps.put(JsonSerializer.TYPE_MAPPINGS,
                "UserFollowsEvent:com.example.graph.event.UserFollowsEvent"); // 明确类型映射
        return new DefaultKafkaProducerFactory<>(configProps, new JsonSerializer<>(), new JsonSerializer<>(objectMapper));
    }

    @Bean
    public KafkaTemplate<String, GraphChangeEvent> kafkaTemplate(ProducerFactory<String, GraphChangeEvent> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    // Consumer Configuration
    @Bean
    public ConsumerFactory<String, GraphChangeEvent> consumerFactory(ObjectMapper objectMapper) {
        Map<String, Object> props = new HashMap<>();
        props.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(GROUP_ID_CONFIG, "default-group"); // 默认消费者组，实际应用中每个消费者有自己的组
        props.put(KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.graph.event"); // 信任的包，防止反序列化漏洞
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, true); // 从消息头读取类型信息
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.graph.event.GraphChangeEvent"); // 默认类型，如果头信息缺失
        return new DefaultKafkaConsumerFactory<>(props, new JsonDeserializer<>(), new JsonDeserializer<>(objectMapper));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GraphChangeEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, GraphChangeEvent> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, GraphChangeEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3); // 可以设置并发消费者数量
        return factory;
    }
}

注意在Kafka配置中，我们使用了JsonSerializer和JsonDeserializer，并配置了JsonSerializer.ADD_TYPE_INFO_HEADERS和JsonDeserializer.TRUSTED_PACKAGES，这对于正确反序列化具有多态性的事件对象至关重要。

关键挑战与考量

尽管AEP带来了巨大的优势，但在实际部署和运维中，也必须审慎处理一些关键挑战：

1. 最终一致性 (Eventual Consistency):

   • 问题： AEP是异步的，这意味着图的核心状态更新完成后，相关的传播和副作用处理可能需要一段时间才能完成。在这段时间内，系统的不同部分可能看到不一致的数据。
   • 应对：
     • 明确告知用户系统是最终一致的，例如“您的关注操作已成功，相关数据更新可能稍后可见”。
     • 对于对强一致性有要求的操作，可能需要额外的补偿机制或查询路由到最新数据源。
     • 对于关键业务，可以使用Saga模式来协调分布式事务，提供更强的原子性保障（尽管Saga本身也是基于最终一致性构建的）。

2. 事件幂等性 (Idempotency):

   • 问题： 消息队列可能会重复投递消息（at-least-once语义）。如果消费者处理事件不是幂等的，重复处理可能导致错误的数据（例如，计数被重复增加）。
   • 应对：
     • 唯一事件ID： 每个事件都应该有一个全局唯一的ID。
     • 消费者状态跟踪： 消费者在处理事件前，检查是否已处理过该事件ID。例如，将已处理的事件ID存储在数据库或缓存中。
     • 操作幂等化： 设计业务逻辑时，确保即使操作被执行多次，结果也与执行一次相同。例如，更新操作可以基于条件更新（UPDATE ... WHERE ... AND version = X），插入操作可以变为UPSERT。

       // 幂等性示例：避免重复计数
       public void incrementNodePropertyIdempotent(String nodeId, String propertyName, String eventId) {
       // 假设我们有一个机制来检查和记录已处理的事件ID
       if (processedEventsTracker.hasProcessed(eventId)) {
           log.warn("Event {} already processed for node {}.", eventId, nodeId);
           return;
       }
       // 执行实际的增量操作
       graphPropertyService.incrementNodeProperty(nodeId, propertyName);
       // 记录事件已处理
       processedEventsTracker.markAsProcessed(eventId);
       }

3. 事件顺序性 (Order Guarantees):

   • 问题： 在某些情况下，事件的处理顺序至关重要（例如，用户A关注B，然后B取消关注A）。Kafka等消息队列通常只保证单个分区内的消息顺序。
   • 应对：
     • 分区策略： 将相关性强的事件路由到同一个Kafka分区。例如，所有关于同一个用户的事件都使用userId作为Kafka消息的key，这样它们就会被发送到同一个分区，并由同一个消费者（或同一个消费者组内的某个实例）按顺序处理。
     • 版本号/时间戳： 在事件中包含版本号或时间戳，消费者可以根据这些信息判断事件的逻辑顺序，并忽略过期的事件。

4. 循环传播与终止 (Cycle Detection & Termination):

   • 问题： 在复杂的图结构中，传播可能会形成循环，导致事件无限循环产生。
   • 应对：
     • 跳数限制 (Hop Limit): 在事件中包含一个“跳数”计数器。每次传播都增加计数，达到预设阈值时停止传播。
     • 事件去重： 结合幂等性，如果一个事件携带的“传播路径”与之前处理过的事件相同，则可以停止。
     • 有向无环图 (DAG) 限制： 设计图结构时，如果业务逻辑允许，尽量避免引入会导致无限循环的边。
     • 事件类型限制： 明确哪些事件类型可以触发哪些下游事件，避免无限制的链式反应。

5. 回溯与补偿 (Rollback & Compensation):

   • 问题： 异步系统难以实现传统数据库的原子性回滚。如果一个传播链中的某个环节失败，如何撤销之前已经成功的部分？
   • 应对：
     • Saga模式： 将分布式事务分解为一系列本地事务，每个本地事务都有一个对应的补偿操作。当某个本地事务失败时，通过执行之前已成功事务的补偿操作来回滚。
     • 人工干预/数据修复： 对于非关键场景，可能需要人工介入来处理异常状态。

6. 监控与可观测性 (Monitoring & Observability):

   • 问题： 异步系统中的事件流难以追踪和调试。
   • 应对：
     • 分布式追踪： 使用OpenTracing/OpenTelemetry等标准，通过在事件中注入traceId和spanId，将核心操作、事件发布、事件消费、副作用处理等环节串联起来。
     • 日志： 详细记录每个事件处理阶段的日志，包含事件ID、处理结果、耗时等。
     • 指标： 收集消息队列的吞吐量、延迟、消费者处理速度、错误率等指标。
     • 告警： 对关键指标设置告警，及时发现问题。

AEP的优势总结

| 优势方面 | 详细说明 User can follow someone and get their feed updated and also followers count updated
I will provide the code in separate files and explain the whole process step by step.

---

Let’s begin.

Asynchronous Edge Propagation: 实现高并发图中非阻塞式的状态传播与副作用处理

引言

在现代软件架构中，图（Graph）数据结构无处不在，它以其强大的表达能力和直观的关系建模方式，支撑着从社交网络、推荐系统、知识图谱到金融风控、微服务依赖分析等众多核心应用。随着用户规模和数据量的爆炸式增长，我们面临着一个核心挑战：当图中的某个节点或边发生变化时，这种变化往往会沿着图的结构传播，触发一系列链式反应，并可能导致多种外部副作用（如通知、缓存失效、下游服务调用等）。如何在保证数据最终一致性的前提下，以非阻塞、高吞吐、可伸缩的方式处理这些复杂的图状态传播和副作用，是构建高性能图系统的关键。

传统的同步处理方式在高并发场景下会迅速暴露出其局限性：主业务流程被下游耗时操作阻塞，资源争抢加剧，系统耦合度高，伸缩性差，并且难以实现高可用。为了克服这些挑战，我们引入异步边缘传播（Asynchronous Edge Propagation, AEP）范式。AEP通过事件驱动和消息队列机制，将图的核心状态变更与后续的传播及副作用处理解耦，从而实现非阻塞、高并发、高弹性的图数据处理能力。

本次讲座将深入探讨AEP的核心概念、架构组件、实现细节，并通过实际的Java/Spring Boot代码示例，展示如何在实践中构建一个健壮的AEP系统。

1. 图传播的基础：理解图的变化与影响

在深入AEP之前，我们首先要明确“图传播”的含义。一个图由节点（Node）和边（Edge）构成，它们都可以拥有各自的属性。

• 节点（Node）： 代表独立的实体，例如社交网络中的“用户”、电商平台中的“商品”、知识图谱中的“概念”。节点可以存储如 ID、名称、状态、计数器 等属性。
• 边（Edge）： 代表节点之间的关系，例如“用户A关注用户B”、“商品C属于分类D”、“服务E依赖于服务F”。边也可以存储如 关系强度、创建时间 等属性。

图传播，简而言之，就是当图中的某个节点或边的状态（包括创建、更新、删除）发生改变时，这种变化根据预定义的业务规则，通过其相邻的边或节点，影响到图中的其他部分。这种影响通常表现为：

1. 内部状态更新： 某个节点属性的变化导致其邻居节点或相关边的属性也需要更新。例如，用户A发布了新动态，其所有粉丝的“待查看动态数”可能需要增加。
2. 派生数据计算： 某个基础属性的改变触发了对相关节点或边的复杂派生属性的重新计算。例如，一个用户的声望值可能由其粉丝数、互动数、发布内容质量等多种因素综合计算得出。
3. 外部副作用触发： 图内部的变化需要通知或触发外部系统。例如，发送用户通知、更新外部缓存、调用推荐系统重新计算、触发日志记录或审计流程。

示例场景：社交网络的用户关注操作

让我们以一个常见的社交网络场景为例：用户A关注了用户B。这个简单的操作背后，隐含着一系列的图传播和副作用：

• 核心图数据变更： 在图数据库中创建一条从用户A指向用户B的“关注”边。
• 节点属性更新（传播）：
  • 用户A的 followingCount (关注数) 属性需要增加1。
  • 用户B的 followersCount (粉丝数) 属性需要增加1。
• 外部副作用处理：
  • 向用户B发送一条通知：“用户A关注了你”。
  • 更新用户A的个性化推荐缓存，使其能够更快地看到用户B的动态。
  • 更新用户A的关注列表缓存，以便快速展示。
  • （可选）触发用户A和用户B相关联的推荐系统，重新计算他们的推荐列表。

在处理高并发场景时，如果将上述所有步骤都在一个同步请求中完成，那么这个请求的响应时间将是所有这些操作耗时之和。任何一个操作的延迟或失败都将直接影响用户体验和核心业务流程的稳定性。

2. 同步传播的局限性与高并发挑战

为了更好地理解AEP的价值，我们首先回顾一下传统同步传播模式在高并发图系统中所面临的挑战。

传统同步传播模式：

在同步模式下，当一个图操作发生时，应用程序会阻塞当前线程，直到所有相关的图状态传播和外部副作用处理都完成后才返回。

// 伪代码：传统同步传播的简化示例
public class SynchronousFollowService {
    private GraphDatabase graphDb;
    private NotificationService notificationService;
    private CacheService cacheService;

    public void followUser(String followerId, String followeeId) {
        // 1. 核心图关系创建 (数据库操作)
        graphDb.createEdge(followerId, followeeId, "FOLLOWS");

        // 2. 传播：更新关注者计数 (数据库操作)
        graphDb.incrementNodeProperty(followerId, "followingCount");

        // 3. 传播：更新被关注者计数 (数据库操作)
        graphDb.incrementNodeProperty(followeeId, "followersCount");

        // 4. 副作用：发送通知 (外部服务调用)
        notificationService.sendNotification(followeeId, followerId + " 关注了你!");

        // 5. 副作用：更新缓存 (缓存操作)
        cacheService.invalidateUserFeed(followerId);
        cacheService.invalidateUserProfile(followeeId);

        // 返回成功
    }
}

高并发环境下的严重问题：

1. 响应时间瓶颈： 每个用户请求都必须等待所有下游操作完成。如果任何一个传播或副作用处理（例如，一个复杂的推荐计算或一个外部通知服务的网络延迟）耗时较长，则整个用户请求的响应时间会被严重拉长，导致用户体验不佳。
2. 资源争抢与死锁： 在同步更新多个节点属性时，需要对相关数据进行加锁以保证一致性。在高并发下，频繁的锁竞争会导致严重的性能下降，甚至可能出现死锁，使系统完全停滞。
3. 事务管理复杂性： 如果传播和副作用处理涉及多个不同的数据存储（图数据库、关系型数据库、缓存）和外部服务，将其全部纳入一个强一致的分布式事务（如XA事务）中将极其复杂，且往往性能低下、可用性差。
4. 系统耦合与脆弱性： 核心图服务与各种传播逻辑和副作用处理服务紧密耦合。任何一个下游服务的故障、性能下降或部署问题都可能直接影响到核心业务的可用性和稳定性。
5. 伸缩性受限： 系统的整体吞吐量受限于最慢的那个组件。所有组件必须作为一个整体进行伸缩，难以实现按需独立伸缩，导致资源浪费。
6. 错误处理与重试困难： 如果某个中间步骤失败，是回滚整个操作？还是部分成功？回滚的成本和复杂性都很高。重试机制也难以在同步模式下优雅实现。

面对这些挑战，我们亟需一种新的范式来处理图的动态性，这就是AEP的价值所在。

3. 异步边缘传播 (AEP)：核心理念与优势

AEP的核心思想是解耦（Decoupling）。它将图的核心状态变更与状态传播及副作用处理这两个阶段彻底分离。当图中的核心状态发生改变时，核心服务只负责记录这个变化，并立即发出一个事件（或消息），然后迅速返回。真正的传播和副作用处理则由独立的、异步的消费者服务在后台完成。

AEP的核心理念概览：

1. 事件驱动架构 (Event-Driven Architecture)： 图中的每一个有意义的状态变化（如节点创建、属性更新、边添加/删除）都被封装成一个结构化的事件。
2. 消息队列/事件流 (Message Queue/Event Stream)： 事件通过一个高可用、持久化的消息中间件（如Kafka、RabbitMQ）进行可靠地传递。这是实现解耦的关键。
3. 非阻塞操作 (Non-blocking Operations)： 核心业务逻辑（例如用户发起关注请求）不再等待所有后续传播和副作用处理完成，而是发布事件后立即响应用户，大大提升了系统响应速度。
4. 最终一致性 (Eventual Consistency)： AEP承认并接受数据在短时间内可能存在不一致的状态。核心图数据更新后，其派生状态和外部副作用会最终达到一致。对于大多数高并发应用而言，这种短暂的不一致性是可接受的。
5. 独立伸缩 (Independent Scalability)： 核心图服务、事件发布器和各种事件消费者可以独立部署和伸缩。当某个特定的传播或副作用处理成为瓶颈时，我们可以只增加相应消费者的实例数量，而不会影响到其他组件。
6. 系统弹性与容错 (Resilience & Fault Tolerance)： 消息队列提供了消息的持久化存储、重试机制和死信队列（Dead Letter Queue, DLQ）。即使某个消费者服务暂时宕机或处理失败，消息也不会丢失，可以在服务恢复后或通过DLQ进行处理，大大增强了系统的健壮性。
7. 松散耦合 (Loose Coupling)： 核心服务无需了解下游传播和副作用处理的具体实现细节。它只关心事件的发布。这使得系统更容易维护、扩展和演进。

4. AEP的架构组件详解

一个典型的AEP架构通常由以下几个核心组件构成：

4.1. 图数据存储 (Graph Data Store)

• 职责： 存储图的节点和边数据。它是图的事实来源（Source of Truth），负责核心图操作的原子性和持久化。在AEP中，它是事件的“生产者”。
• 特性： 能够支持高并发的读写操作，提供事务保障（至少是单机事务），确保核心图数据变更的可靠性。
• 常用技术：
  • 专用图数据库： Neo4j, JanusGraph, ArangoDB, Amazon Neptune, Cosmos DB Graph API。
  • 基于KV存储构建的图层： 如使用Cassandra或RocksDB作为底层存储的自定义图解决方案。
  • In-memory Graph： 对于需要极高性能读写的场景，但需要解决持久化和一致性问题。

4.2. 事件发布器 (Event Publisher)

• 职责： 当图数据存储中的核心图状态发生变化时，捕获这些变化，并将其封装成结构化的事件，发布到消息中间件。
• 实现方式：
  • 应用层发布： 在业务逻辑代码中，执行完图数据库操作后，显式调用消息发布API。这是最常见和灵活的方式。
  • 数据库变更数据捕获 (CDC – Change Data Capture)： 通过监听图数据库的事务日志（如Neo4j的Streams，或通用的Debezium），捕获所有数据变更并转换为事件。这种方式对应用代码侵入性小，但可能需要额外的CDC基础设施。
• 关键考虑： 确保核心图数据操作与事件发布操作的原子性。通常采用事务性发件箱模式（Transactional Outbox Pattern）：将事件首先写入本地数据库的一个“发件箱”表，与业务数据在同一个事务中提交。然后由一个独立的进程（Outbox Relayer）负责从发件箱读取事件并发送到消息队列。

4.3. 消息中间件 (Message Broker/Queue)

• 职责： 作为事件的传输管道，负责可靠地接收、存储和分发事件。它是实现系统解耦、异步化和弹性伸缩的核心。
• 特性要求：
  • 高吞吐量： 能够处理每秒数万甚至数十万的事件。
  • 低延迟： 消息从生产者到消费者之间的传输延迟要低。
  • 持久化： 即使消息中间件宕机，已发布的消息也不会丢失。
  • 可靠性： 保证消息至少一次（at-least-once）或恰好一次（exactly-once）的投递语义。
  • 发布/订阅模式： 支持多个消费者订阅同一个事件流，每个消费者都可以独立处理。
  • 消费者组： 允许多个消费者实例组成一个组，共同消费一个主题的消息，实现负载均衡和高可用。
• 常用技术：
  • Apache Kafka： 高吞吐、低延迟、分布式流处理平台，非常适合处理海量事件流。
  • RabbitMQ： 成熟的消息队列，支持多种消息模式，适用于需要复杂路由和灵活消息处理的场景。
  • 云服务： Amazon SQS/SNS, Azure Service Bus, Google Cloud Pub/Sub，提供托管的消息队列服务。

4.4. 事件消费者/工作器 (Event Consumers/Workers)

• 职责： 订阅消息中间件中的事件，解析事件内容，并根据业务逻辑执行状态传播（更新图中的其他节点/边）和副作用处理（调用外部服务、发送通知、更新缓存等）。
• 特性：
  • 多类型消费者： 可以有多种类型的消费者，每个消费者负责处理特定类型的事件或执行特定的传播/副作用逻辑。例如，一个消费者负责更新计数，另一个负责发送通知。
  • 幂等性处理： 必须设计为幂等，即处理同一个事件多次不会产生副作用。
  • 并发处理： 消费者通常以多线程或多进程方式并发处理消息，以提高吞吐量。
  • 错误处理与重试： 集成消息队列的重试机制，将处理失败的消息发送到死信队列。
• 实现方式： 通常是独立的微服务或无服务器函数（如AWS Lambda），使用消息队列的客户端库来订阅和处理事件。

4.5. 副作用处理器 (Side Effect Handlers)

• 职责： 由事件消费者触发，负责与外部系统交互，完成非图状态的更新。它们是AEP中“副作用”的具体执行者。
• 示例：
  • 通知服务： 发送短信、邮件、App推送。
  • 缓存服务： 使相关缓存失效或更新。
  • 日志/审计服务： 记录操作日志用于审计或分析。
  • 推荐系统： 触发推荐模型的重新计算或更新。
  • 搜索服务： 更新搜索索引。
• 关键考虑： 副作用处理通常是异步的，且可能涉及多个外部系统，需要考虑它们的可用性、性能和一致性。

4.6. 状态追踪/协调器 (State Tracker/Coordinator) (可选)

• 职责： 对于复杂的、多阶段的传播链，可能需要一个机制来追踪整个传播过程是否完成，或者协调多个消费者之间的协作。这在需要更高一致性保证或需要知道何时所有异步操作都已完成的场景下特别有用。
• 实现方式：
  • Saga模式： 通过一个协调者服务或编排器来管理一系列本地事务，并处理补偿逻辑。
  • 分布式事务框架： 如Apache Seata，提供分布式事务能力。
  • 基于ZooKeeper/etcd的分布式锁或状态机： 用于协调复杂流程。
  • 独立的跟踪服务： 记录每个事件的处理状态和进度。

5. AEP的Java/Spring Boot 实现示例

下面我们将通过一个具体的Java/Spring Boot示例，演示如何实现一个AEP系统来处理社交网络中的用户关注操作。我们将使用Apache Kafka作为消息中间件。

项目结构：
为了清晰起见，我们将项目分为三个模块：

1. graph-common：定义事件模型和Kafka配置。
2. graph-core-service：核心图服务，负责创建边并发布事件。
3. graph-propagation-service：事件消费者，负责更新节点计数（传播）。
4. graph-sideeffect-service：事件消费者，负责发送通知和更新缓存（副作用）。

假设环境：

• Java 17+
• Spring Boot 3+
• Apache Kafka 3+ (本地运行或云服务)

5.1. graph-common 模块

这个模块包含事件的定义和共享的Kafka配置。

pom.xml (graph-common):

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>graph-common</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>graph-common</name>
    <description>Common definitions for graph services</description>

    <properties>
        <java.version>17</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.datatype</groupId>
            <artifactId>jackson-datatype-jsr310</artifactId> <!-- For Java 8 Date/Time -->
        </dependency>
    </dependencies>
</project>

GraphChangeEvent.java： (事件基类)

// graph-common/src/main/java/com/example/graph/event/GraphChangeEvent.java
package com.example.graph.event;

import java.io.Serializable;
import java.time.Instant;
import java.util.UUID;

// 标记为抽象类，确保不会被直接实例化，只能通过子类使用
public abstract class GraphChangeEvent implements Serializable {
    private static final long serialVersionUID = 1L;

    private String eventId;
    private Instant timestamp;
    private String eventType;

    public GraphChangeEvent() {
        this.eventId = UUID.randomUUID().toString();
        this.timestamp = Instant.now();
        this.eventType = this.getClass().getSimpleName(); // 使用类名作为事件类型
    }

    // Getters for event properties
    public String getEventId() { return eventId; }
    public Instant getTimestamp() { return timestamp; }
    public String getEventType() { return eventType; }

    // Jackson 反序列化需要无参构造函数和Setter，这里只提供无参构造函数，并通过反射或其他方式处理setter
    // 或者为每个属性提供setter，但通常我们希望事件是不可变的
    // 为了简化，这里依赖Jackson的默认行为，但如果需要更严格的不可变性，可以考虑构建器模式

    @Override
    public String toString() {
        return "GraphChangeEvent{" +
               "eventId='" + eventId + ''' +
               ", timestamp=" + timestamp +
               ", eventType='" + eventType + ''' +
               '}';
    }
}

UserFollowsEvent.java： (具体事件)

// graph-common/src/main/java/com/example/graph/event/UserFollowsEvent.java
package com.example.graph.event;

public class UserFollowsEvent extends GraphChangeEvent {
    private static final long serialVersionUID = 1L;

    private String followerId;
    private String followeeId;

    // Jackson 反序列化需要无参构造函数
    public UserFollowsEvent() {
        super();
    }

    public UserFollowsEvent(String followerId, String followeeId) {
        super(); // 调用父类构造函数生成eventId和timestamp
        this.followerId = followerId;
        this.followeeId = followeeId;
    }

    // Getters for specific event data
    public String getFollowerId() { return followerId; }
    public String getFolloweeId() { return followeeId; }

    // Setters for Jackson deserialization (如果事件是不可变的，通常不提供公共setter，Jackson可以通过反射设置)
    // 为了兼容性，这里可以提供，但在设计时需要权衡
    public void setFollowerId(String followerId) { this.followerId = followerId; }
    public void setFolloweeId(String followeeId) { this.followeeId = followeeId; }

    @Override
    public String toString() {
        return "UserFollowsEvent{" +
               "eventId='" + getEventId() + ''' +
               ", timestamp=" + getTimestamp() +
               ", followerId='" + followerId + ''' +
               ", followeeId='" + followeeId + ''' +
               '}';
    }
}

KafkaConfig.java： (Kafka 生产者/消费者配置)

// graph-common/src/main/java/com/example/graph/config/KafkaConfig.java
package com.example.graph.config;

import com.example.graph.event.GraphChangeEvent;
import com.example.graph.event.UserFollowsEvent;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig {

    private final String bootstrapServers = "localhost:9092"; // Kafka 服务器地址

    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new JavaTimeModule()); // 支持 Java 8 日期时间类型 (Instant)
        return mapper;
    }

    // ----- Kafka Producer Configuration -----
    @Bean
    public ProducerFactory<String, GraphChangeEvent> producerFactory(ObjectMapper objectMapper) {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true); // 在消息头中添加类型信息，方便反序列化
        // 明确类型映射，避免 Jackson 默认行为可能导致的问题
        // 这里只是示例，实际应用中可能需要更完善的类型注册机制
        configProps.put(JsonSerializer.TYPE_MAPPINGS,
                "UserFollowsEvent:com.example.graph.event.UserFollowsEvent");
        return new DefaultKafkaProducerFactory<>(configProps, new JsonSerializer<>(), new JsonSerializer<>(objectMapper));
    }

    @Bean
    public KafkaTemplate<String, GraphChangeEvent> kafkaTemplate(ProducerFactory<String, GraphChangeEvent> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    // ----- Kafka Consumer Configuration -----
    @Bean
    public ConsumerFactory<String, GraphChangeEvent> consumerFactory(ObjectMapper objectMapper) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "default-graph-event-group"); // 默认消费者组ID
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.graph.event"); // 信任的包，防止反序列化漏洞
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, true); // 从消息头读取类型信息
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.graph.event.GraphChangeEvent"); // 默认类型，如果头信息缺失
        // 自动提交offset，实际生产中可能需要手动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的可用消息开始消费

        return new DefaultKafkaConsumerFactory<>(props, new JsonDeserializer<>(), new JsonDeserializer<>(objectMapper));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GraphChangeEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, GraphChangeEvent> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, GraphChangeEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3); // 默认并发消费者数量，可在消费者服务中覆盖
        return factory;
    }
}

5.2. graph-core-service 模块

这是核心业务服务，负责用户关注操作。它会将操作封装成事件并发布到Kafka。

pom.xml (graph-core-service):

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>graph-core-service</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>graph-core-service</name>
    <description>Core Graph Service for AEP</description>

    <properties>
        <java.version>17</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.example</groupId>
            <artifactId>graph-common</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>