@Aspect
@Component
@Slf4j
public class DomainEventProcessor {

    /**
     * 这里我是我对RocketMq的封装
     */
    @Autowired
    private EventPublisherExecutor processPublisherExecutor;

    /**
     * 当前上下文内订阅者
     */
    @Autowired
    protected ApplicationContext applicationContext;

    private static ThreadLocal<AtomicInteger> counter = new ThreadLocal<>();

    @Pointcut("within(com.github.tom.*.application..*)")
    public void aopRule() {

    }

    /**
     * 为当前线程初始化EventBus
     */
    @Before("aopRule()")
    public void initEventBus(JoinPoint joinPoint) {
        log.debug("初始化领域事件槽");
        log.debug("切入切点信息：" + joinPoint.getSignature().toString());
        EventBus.instance().init();
        if (counter.get() == null) {
            counter.set(new AtomicInteger(0));
        }
        counter.get().incrementAndGet();
    }

    /**
     * 发布领域事件
     */
    @AfterReturning("aopRule()")
    public void publish() {
        int count = counter.get().decrementAndGet();
        if (count == 0) {
            try {
                List<DomainEvent> domainEventList = EventBus.instance().getDomainEvent();
                if (domainEventList != null && domainEventList.size() > 0) {
                    //进程内事件
                    domainEventList.forEach(domainEvent -> applicationContext.publishEvent(domainEvent));
                    //进程外事件
                    domainEventList.forEach(domainEvent -> processPublisherExecutor.publish(domainEvent));
                }
            } finally {
                EventBus.instance().reset();
                counter.set(null);
            }
        }
    }

    @AfterThrowing(throwing = "ex", pointcut = "aopRule()")
    public void exception(Throwable ex) {
        log.error(ex.getMessage(), ex);
        EventBus.instance().reset();
        //释放计数器
        counter.set(null);
    }
}

这里借助了AOP功能，在AOP内我对service进行拦截，在执行方法拦截的出口时，查找当前线程内的EventBus中看是否有存在的领域事件，接下来把事件发送出去，事件的响应分为进程内和进程外（多微服务），刚才的同学问的如何持久化到DB这里可以看到答案

@Slf4j
public abstract class AbstractEventHandler<T extends EventData> implements SmartApplicationListener {

    private Class<?> clazzType;

    public AbstractEventHandler(Class<? extends ApplicationEvent> clazzType) {
        this.clazzType = clazzType;
    }

    @Override
    public boolean supportsEventType(Class<? extends ApplicationEvent> clazzType) {
        return clazzType == this.clazzType;
    }

    @Override
    public boolean supportsSourceType(Class<?> clazz) {
        return true;
    }

    @Override
    public void onApplicationEvent(ApplicationEvent applicationEvent) {
        onApplicationEventHandler((T) applicationEvent);
    }

    protected abstract void onApplicationEventHandler(T eventData);
}


@Slf4j
public abstract class AbstractPersistenceEventHandler<T extends EventData> extends AbstractEventHandler<T> {

    public AbstractPersistenceEventHandler(Class<? extends ApplicationEvent> clazzType) {
        super(clazzType);
    }

    @Override
    public int getOrder() {
        return 0;
    }

}

@Component
public class ChangeProductStockPersistenceEventHandler
        extends AbstractPersistenceEventHandler<ChangedProductStockDomainEvent> {

    @Autowired
    private ProductRepository productRepository;

    public CreatedPortalArticlePersistenceEventHandler() {
        super(CreatedPortalArticleDomainEvent.class);
    }

    @Override
    protected void onApplicationEventHandler(ChangedProductStockDomainEvent eventData) {
        if (portalArticleRepository.updateById(eventData.getProduct()) <= 0) {
            throw new BusinessException("数据操作错误");
        }
    }
}

@Component
public class JdStockEventHandler {

    @Autowired
    private JdAppService jdAppService;

    /**
     * 库存持久化事件
     *
     * @param eventData
     */
    @StreamListener(value = "product-channel")
    public void receive(@Payload ChangedProductStockDomainEvent eventData) {
        jdAppService.changingInventory(eventData);
    }
}

@Component
public class ProductStockTest {

    @Before
    public void setUp() {
      EventBus.instance().init();
    }
    
    @Test
    public void testChangeStockError() {
        Product product = new Product();
        product.setStock(BigDecimal.valueOf("-1"));
        product.changeStock();
    }

   @Test
    public void testChangeStockSuccess() {
        Product product = new Product();
        product.setStock(BigDecimal.valueOf("2"));
        product.changeStock();
        assertThat(product.getStock()).isEqualTo("2");
    }

}