扩展点

soda 的每一个关键 Bean 都带着 @ConditionalOnMissingBean。这意味着：你自己定义一个同名 Bean，框架的就让位。 这一页讲每个扩展口怎么用。

扩展点一览：

想换掉什么	定义这个 Bean	位置
事件总线	`EventBus`	`com.hibuka.soda.cqrs.event`
事件序列化器	`EventSerializer`	`com.hibuka.soda.event`
命令总线	`CommandBus`	`com.hibuka.soda.cqrs.command`
查询总线	`QueryBus`	`com.hibuka.soda.cqrs.query`
门面	`BusFacade`	`com.hibuka.soda.bus.facade`
幂等服务	`IdempotencyService`	`com.hibuka.soda.event.redis.service`
写仓储	实现 `WriteRepository<T>`	`com.hibuka.soda.domain.repository`

自定义事件总线

EventBus 接口只有三个方法：

java

public interface EventBus {
    void publish(DomainEvent event) throws BaseException;
    void subscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
    void unsubscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
}

要换成自己的实现（比如接 Kafka、接 Pulsar、或者只做审计日志），定义一个 Bean：

java

package com.example.demo.infrastructure.eventbus;

import com.hibuka.soda.cqrs.event.EventBus;
import com.hibuka.soda.cqrs.event.EventHandler;
import com.hibuka.soda.domain.event.DomainEvent;
import com.hibuka.soda.foundation.error.BaseException;
import org.springframework.stereotype.Component;

@Component
public class KafkaEventBus implements EventBus {

    @Override
    public void publish(DomainEvent event) throws BaseException {
        // 发到 Kafka
    }

    @Override
    public void subscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException {
        // 注册订阅
    }

    @Override
    public void unsubscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException {
        // 取消订阅
    }
}

因为 SimpleEventBusAutoConfiguration 和 RedisEventBusAutoConfiguration 里的 Bean 都是 @ConditionalOnMissingBean，你定义了就自动生效，框架不会再创建 SimpleEventBus。

技巧

但要小心 RedisEventBusAutoConfiguration 里的 redisStreamEventBus 是 @Primary 的。 如果你的 KafkaEventBus 和它同时存在，注入 EventBus 时会拿到 @Primary 的那个。你的 Bean 想生效，要么不要引入 soda-event-redis 模块，要么给 @Primary 标到你的 Bean 上，或者用 @Qualifier 精确注入。

自定义总线的三个义务：

必须能反序列化事件。 跨进程的总线需要知道事件的类型和内容。
必须尊重事务边界。 参考 RedisStreamEventBus.publish 的写法：有活动事务时注册 afterCommit，没有时立即发。
必须容错。 参考 RepositoryEventAspect.publishEvents 的写法：单个事件发布失败只记日志，不影响其它事件。

自定义事件序列化器

soda-core-base 里的 com.hibuka.soda.event.EventSerializer：

java

public interface EventSerializer {
    String toJson(Object obj);

    <T> T fromJson(String json, Class<T> type);

    class SerializationException extends RuntimeException {
        public SerializationException(String message, Throwable cause) {
            super(message, cause);
        }
    }
}

这个抽象的作用是把事件序列化从具体 JSON 库里解耦。默认实现是 JacksonEventSerializer（在 soda-event-redis 里），用反射调用 ObjectMapper 的 writeValueAsString / readValue。

要换成 fastjson、Gson、Kryo 或者自己的二进制协议：

java

package com.example.demo.infrastructure.event;

import com.alibaba.fastjson.JSON;
import com.hibuka.soda.event.EventSerializer;
import org.springframework.stereotype.Component;

@Component
public class FastjsonEventSerializer implements EventSerializer {

    @Override
    public String toJson(Object obj) {
        try {
            return JSON.toJSONString(obj);
        } catch (Exception e) {
            throw new SerializationException("Failed to serialize event", e);
        }
    }

    @Override
    public <T> T fromJson(String json, Class<T> type) {
        try {
            return JSON.parseObject(json, type);
        } catch (Exception e) {
            throw new SerializationException("Failed to deserialize event", e);
        }
    }
}

只要注册成 EventSerializer Bean，RedisEventBusAutoConfiguration 里的默认实现就会让位：

java

@Bean
@ConditionalOnMissingBean
public EventSerializer eventSerializer(ApplicationContext applicationContext) {
    return new JacksonEventSerializer(applicationContext);
}

注意

换序列化器时要确保收发两端用的是同一个格式。如果生产方用 fastjson、消费方用 Jackson，跨服务的反序列化会失败。事件类上的 @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "@class") 是给 Jackson 用的 —— 换成别的库时，@class 这个字段的处理方式要自己实现。

双 Jackson 支持是怎么做到的

这是 17.0.1 最主要的技术变化：同一个 jar 同时支持 Spring Boot 3.2.x（Jackson 2）和 Spring Boot 4.x（Jackson 3）。

问题的根源：Jackson 3 的包名从 com.fasterxml.jackson.* 变成了 tools.jackson.*。如果 soda-core 直接 import ObjectMapper，就不可能同时兼容两个版本。

解法是反射。JacksonEventSerializer 里没有一行 Jackson 的 import：

java

public class JacksonEventSerializer implements EventSerializer {

    private final ApplicationContext applicationContext;
    private volatile Object objectMapper;
    private volatile Method writeValueAsString;
    private volatile Method readValue;

    public JacksonEventSerializer(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    private Object getObjectMapper() {
        if (objectMapper == null) {
            synchronized (this) {
                if (objectMapper == null) {
                    objectMapper = resolveObjectMapper();
                    try {
                        this.writeValueAsString = objectMapper.getClass().getMethod("writeValueAsString", Object.class);
                        this.readValue = objectMapper.getClass().getMethod("readValue", String.class, Class.class);
                    } catch (NoSuchMethodException e) {
                        throw new SerializationException("ObjectMapper missing required methods", e);
                    }
                }
            }
        }
        return objectMapper;
    }
    ...
}

ObjectMapper 是从 Spring 容器里按名字找的：

java

private Object resolveObjectMapper() {
    String[] names = {"jacksonObjectMapper", "jacksonJsonMapper", "jsonMapper", "objectMapper"};
    for (String name : names) {
        try {
            if (applicationContext.containsBean(name)) {
                return applicationContext.getBean(name);
            }
        } catch (Exception ignored) {}
    }
    // Fallback: scan by class name
    for (String beanName : applicationContext.getBeanDefinitionNames()) {
        try {
            Object bean = applicationContext.getBean(beanName);
            String className = bean.getClass().getName();
            if (className.contains("ObjectMapper") || className.contains("JsonMapper")) {
                return bean;
            }
        } catch (Exception ignored) {}
    }
    throw new IllegalStateException("No ObjectMapper bean found. Ensure Jackson is on the classpath.");
}

查找顺序：

按名字找：jacksonObjectMapper → jacksonJsonMapper → jsonMapper → objectMapper
名字找不到就遍历所有 Bean，类名里含 ObjectMapper 或 JsonMapper 的算命中
都没有就抛 IllegalStateException("No ObjectMapper bean found. Ensure Jackson is on the classpath.")

ObjectMapper 是懒加载的 —— 第一次序列化时才解析，而不是在 Bean 创建时。这样 Spring Boot 的 Jackson 自动配置有时间完成。

技巧

因为是按类名扫的兜底匹配，你的容器里最好只有一个 ObjectMapper。如果有多个（比如一个给 Web 用，一个给 Redis 用），命中哪个是不确定的 —— 这时候你应该显式定义一个 EventSerializer Bean，或者给期望的那个 ObjectMapper 起个上面列出的名字。

提示

soda-core-base 本身不依赖 Jackson。Jackson 只在 soda-event-redis 里通过 EventSerializer 抽象间接使用。所以只用进程内事件总线的工程，classpath 上没有 Jackson 也完全没问题。

自定义写仓储

写仓储不需要「扩展点」——它本来就是给你实现的接口：

java

public interface WriteRepository<T extends AbstractAggregateRoot> {
    boolean save(T aggregate) throws BaseException;
    boolean update(T aggregate) throws BaseException;
    boolean delete(T aggregate) throws BaseException;
    boolean operate(T aggregate, int operationType) throws BaseException;
}

实现它，把实现类交给 Spring 管理就行。要守住两件事：仓储类型（接口或实现类）的名字里有 Repository，方法名是这四个 —— RepositoryEventAspect 的切点 *..*Repository+ 匹配的就是这两样。实现类作为接口的子类型，无论叫什么都会被拦到。

自定义幂等服务

幂等是 Redis 事件总线的特性。IdempotencyService 定义在 com.hibuka.soda.event.redis.service：

java

public interface IdempotencyService {
    enum ProcessingStatus {
        PROCESSING, SUCCESS, FAILED
    }

    boolean beginProcessing(String eventId);

    void markAsSuccess(String eventId, Map<String, Object> handlerResults);

    void markAsFailed(String eventId, String error);

    ProcessingStatus getStatus(String eventId);

    void cleanupExpiredStatus();
}

默认实现是 RedisIdempotencyServiceImpl，用 Redis Hash 存状态：

Hash 字段	内容
`status`	`PROCESSING` / `SUCCESS` / `FAILED`
`processedAt`	处理时间戳
`handlerResults`	各 EventHandler 的处理结果
`error`	失败原因

Redis key 是 idempotency.redisKeyPrefix + ":" + eventId，默认前缀 soda-events-idempotency，默认过期 24 小时。

提示

RedisStreamEventBus 是直接 new 出 RedisIdempotencyServiceImpl 的，不是从容器里注入：

java

this.idempotencyService = new RedisIdempotencyServiceImpl(redisTemplate, idempotencyProperties);

所以你定义一个 IdempotencyService Bean 不会自动替换它。要换实现，得自己继承 RedisStreamEventBus 或自己实现 EventBus。

幂等逻辑在消费端：

java

IdempotencyService.ProcessingStatus status = idempotencyService.getStatus(eventId);
if (status == IdempotencyService.ProcessingStatus.SUCCESS) {
    // 已成功处理过，跳过
} else if (status == IdempotencyService.ProcessingStatus.PROCESSING) {
    // 正在处理中，跳过
}

beginProcessing 的规则：

状态是 SUCCESS → 返回 false，不处理
状态是 PROCESSING → 返回 false，不处理
状态是 FAILED → 返回 true，可以重试
没有记录 → 写入 PROCESSING 并设置过期时间，返回 true

注意

幂等默认是关闭的：soda.event.redis.stream.idempotency.enabled 默认 false。跨服务场景下建议打开 —— Redis Stream 的「至少一次」投递语义意味着同一个事件可能被消费多次。

扩展线程池

soda-core 的异步线程池由 AsyncConfig 配置，前缀是 scoda.ddd.cqrs.async：

java

@ConfigurationProperties(prefix = "scoda.ddd.cqrs.async")
public class AsyncConfig {
    private int corePoolSize = 8;
    private int maxPoolSize = 16;
    private int queueCapacity = 100;
    private String threadNamePrefix = "cqrs-async-";
}

注意

前缀是 scoda.ddd.cqrs.async，不是 soda.ddd.cqrs.async。README-CN 里写的是后者 —— 那是错的，写了不生效。正确写法：

yaml

scoda:
  ddd:
    cqrs:
      async:
        core-pool-size: 16
        max-pool-size: 32
        queue-capacity: 200
        thread-name-prefix: "cqrs-async-"

线程池本身在 SpringBusAutoConfiguration 里创建：

java

@Bean("cqrsAsyncExecutor")
@ConditionalOnMissingBean(name = "cqrsAsyncExecutor")
public Executor cqrsAsyncExecutor(AsyncConfig asyncConfig) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(asyncConfig.getCorePoolSize());
    executor.setMaxPoolSize(asyncConfig.getMaxPoolSize());
    executor.setQueueCapacity(asyncConfig.getQueueCapacity());
    executor.setThreadNamePrefix(asyncConfig.getThreadNamePrefix());
    executor.setKeepAliveSeconds(60);
    executor.setRejectedExecutionHandler((r, executor1) -> {
        logger.warn("[BusAutoConfiguration] Thread pool is full, rejecting task: {}", r.getClass().getName());
        if (!executor1.isShutdown()) {
            try {
                r.run();
            } catch (Exception e) {
                logger.error("[BusAutoConfiguration] Error executing task in caller thread: {}", e.getMessage(), e);
            }
        }
    });
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.initialize();
    return executor;
}

拒绝策略是在调用者线程执行（CallerRunsPolicy 的思路），所以任务不会丢 —— 但调用线程会被阻塞。生产环境要留意这一点：队列打满时，Web 请求线程会跟着卡住。

技巧

想换自己的线程池，定义一个名为 cqrsAsyncExecutor 的 Executor Bean 就行 —— @ConditionalOnMissingBean(name = "cqrsAsyncExecutor") 会让框架放弃创建。

自定义 BusFacade

BusFacade 也是 @ConditionalOnMissingBean。想改发送逻辑（比如加埋点、改递归策略），可以继承它：

java

package com.example.demo.infrastructure.bus;

import com.hibuka.soda.bus.facade.BusFacade;
import com.hibuka.soda.cqrs.command.CommandBus;
import com.hibuka.soda.cqrs.query.QueryBus;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;

@Component
@Primary
public class MetricsBusFacade extends BusFacade {
    public MetricsBusFacade(CommandBus commandBus, QueryBus queryBus, Executor cqrsAsyncExecutor) {
        super(commandBus, queryBus, cqrsAsyncExecutor);
    }

    @Override
    public <C extends Command<R>, R> R sendTransactCommand(C command) throws BaseException {
        // 加埋点，然后调 super
        return super.sendTransactCommand(command);
    }
}

注意

继承 BusFacade 时，BusFacade 的构造函数是 @RequiredArgsConstructor 生成的，参数是 (CommandBus, QueryBus, Executor) —— 顺序不能变，第三个参数要传带 @Qualifier("cqrsAsyncExecutor") 的那个线程池。

扩展点对照表

扩展点	接口	默认实现	生效条件
事件总线	`EventBus`	`SimpleEventBus` / `RedisStreamEventBus`	`soda.event.bus-type`
序列化器	`EventSerializer`	`JacksonEventSerializer`	`soda.event.bus-type=redis`
命令总线	`CommandBus`	`SimpleCommandBus`	总是
查询总线	`QueryBus`	`SimpleQueryBus`	总是
门面	`BusFacade`	`BusFacade`	总是
幂等	`IdempotencyService`	`RedisIdempotencyServiceImpl`	Redis 模式，且不可替换
写仓储	`WriteRepository<T>`	无	你自己实现
读仓储	无接口	无	你自己定义

下一步

完整的配置项清单：配置参考
自动装配是怎么被覆盖的：事件总线与切面
版本与兼容性：版本与兼容

扩展点#

自定义事件总线#

自定义事件序列化器#

双 Jackson 支持是怎么做到的#

自定义写仓储#

自定义幂等服务#

扩展线程池#

自定义 BusFacade#

扩展点对照表#

下一步#

扩展点

自定义事件总线

自定义事件序列化器

双 Jackson 支持是怎么做到的

自定义写仓储

自定义幂等服务

扩展线程池

自定义 BusFacade

扩展点对照表

下一步