如何在Axon Saga中使用非Axon Kafka事件？相关问题咨询

AI 小北

2026-7-26

关于Axon Saga处理非Axon微服务Kafka事件的方案

1. extension-kafka是否能提供帮助？

完全可以。Axon的extension-kafka就是为打通Axon框架与Kafka生态设计的，它支持从Kafka消费消息并转换为Axon可识别的EventMessage——不管消息来自Axon服务还是非Axon服务，只要实现合适的消息转换逻辑，就能被Saga正常处理。

2. 除自定义`KafkaMessageConverter`外的核心困难

事件元数据对齐：Axon的事件（尤其是Saga依赖的）需要特定元数据（如message-id、correlation-id、timestamp）保证追踪、关联和可靠性。非Axon服务发送的Kafka消息通常不会包含这些元数据，或格式不匹配，需在转换时补全、映射这些字段，否则Saga可能无法正确关联实例或处理重试。
事件类型映射：Axon依赖事件类的全限定名区分事件类型，非Axon服务可能用自定义标识（如header中的event-type字段）标记事件类型，需在转换器中将该标识映射到Axon对应的事件类，否则Axon找不到Saga中对应的处理方法。
Saga关联策略适配：Saga通过@AssociationValue关联实例，若非Axon事件的关联字段（如orderId）名称或格式与Saga期望的不一致，要么在转换器中调整payload/元数据映射，要么自定义AssociationResolver，确保能正确匹配到目标Saga实例。
可靠性机制兼容：Axon默认提供重试、死信队列等容错机制，非Axon服务的Kafka消息需适配这些机制。需配置Kafka Consumer的重试策略、死信处理逻辑，与Axon的事件处理器配置对齐，避免消息丢失或重复处理。
序列化兼容性：非Axon服务可能使用不同的序列化方式（如Protobuf、自定义JSON结构），转换器需确保能将Kafka消息的payload正确反序列化为Axon的事件对象，可能需要额外配置序列化器或在转换时做格式转换。

3. 配置示例

自定义KafkaMessageConverter

@Component
public class CustomNonAxonEventConverter extends KafkaMessageConverter {

    private static final Logger log = LoggerFactory.getLogger(CustomNonAxonEventConverter.class);
    private final ObjectMapper objectMapper;

    public CustomNonAxonEventConverter(ObjectMapper objectMapper, EventSerializer eventSerializer) {
        super(eventSerializer);
        this.objectMapper = objectMapper;
    }

    @Override
    protected Optional<EventMessage<?>> convertInbound(KafkaConsumerRecord<?, ?> consumerRecord) {
        // 从Kafka Headers提取事件类型标识
        String eventType = extractHeaderValue(consumerRecord.headers(), "event-type");
        if (eventType == null) {
            log.warn("No event-type header found in Kafka record, skipping");
            return Optional.empty();
        }

        try {
            // 加载对应的Axon事件类
            Class<?> eventClass = Class.forName(eventType);
            // 反序列化消息体为事件对象
            Object payload = objectMapper.readValue(consumerRecord.value().toString(), eventClass);

            // 构建Axon事件元数据，补全必要字段
            MetaData metaData = MetaData.builder()
                    .put("message-id", extractHeaderValue(consumerRecord.headers(), "message-id") != null 
                            ? extractHeaderValue(consumerRecord.headers(), "message-id") 
                            : UUID.randomUUID().toString())
                    .put("timestamp", System.currentTimeMillis())
                    .put("correlation-id", extractHeaderValue(consumerRecord.headers(), "correlation-id") != null
                            ? extractHeaderValue(consumerRecord.headers(), "correlation-id")
                            : UUID.randomUUID().toString())
                    .build();

            // 返回Axon通用事件消息
            return Optional.of(new GenericEventMessage<>(payload, metaData));
        } catch (ClassNotFoundException | JsonProcessingException e) {
            log.error("Failed to convert Kafka record to Axon EventMessage", e);
            return Optional.empty();
        }
    }

    private String extractHeaderValue(Headers headers, String headerKey) {
        Header header = headers.lastHeader(headerKey);
        return header != null ? new String(header.value(), StandardCharsets.UTF_8) : null;
    }
}

Axon Kafka配置类

@Configuration
public class AxonKafkaIntegrationConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Autowired
    private CustomNonAxonEventConverter customNonAxonEventConverter;

    @Bean
    public KafkaMessageSource<String, String> nonAxonEventKafkaSource() {
        // Kafka Consumer配置
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-saga-kafka-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        // 创建Kafka消息源，指定监听的topic和自定义转换器
        return new KafkaMessageSource<>(
                () -> new DefaultKafkaConsumerFactory<>(consumerProps),
                (record, exception) -> {
                    // 死信处理逻辑：发送到DLQ topic
                    log.error("Failed to process record with key: {}, sending to DLQ", record.key(), exception);
                    // 此处可添加发送到DLQ的代码
                },
                customNonAxonEventConverter,
                "external-order-events" // 非Axon服务发送事件的topic
        );
    }

    @Bean
    public EventProcessingConfigurer eventProcessingConfigurer(KafkaMessageSource<String, String> nonAxonEventKafkaSource) {
        return configurer -> configurer.registerSubscribingEventProcessor(
                "order-saga-processor", // 与Saga的@ProcessingGroup对应
                config -> config.messageSource(nonAxonEventKafkaSource)
                        .errorHandler(new PropagatingErrorHandler()) // 可自定义错误处理器
                        .trackingEventProcessorConfiguration(configuration -> configuration
                                .initialSegmentCount(1)
                                .retryScheduler(RetryScheduler.builder().maxRetryCount(3).build()))
        );
    }
}

Saga处理示例

@Saga
@ProcessingGroup("order-saga-processor")
public class OrderProcessingSaga {

    @Autowired
    private transient CommandGateway commandGateway;

    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(ExternalOrderCreatedEvent event) {
        // 关联Saga实例与orderId
        SagaLifecycle.associateWith("orderId", event.getOrderId());
        // 触发支付命令
        commandGateway.send(new InitiatePaymentCommand(event.getOrderId(), event.getTotalAmount()));
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentSuccessfulEvent event) {
        // 触发发货命令
        commandGateway.send(new FulfillOrderCommand(event.getOrderId()));
    }

    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderFulfilledEvent event) {
        // 结束Saga
        log.info("Order {} fulfilled, ending saga", event.getOrderId());
    }
}

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