Event?Sourcing事件溯源模式優(yōu)化業(yè)務(wù)系統(tǒng)

更新時間：2023年07月09日 18:25:41 作者：老鼠AI大米_Java全棧

這篇文章主要為大家介紹了Event?Sourcing事件溯源模式優(yōu)化業(yè)務(wù)系統(tǒng)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

正文

高內(nèi)聚低耦合一直是程序設(shè)計提倡的方式，但是很多人在實際項目中一直再使用面向過程的編碼，導(dǎo)致代碼臃腫不堪，項目維護難度日益增加，在我接觸的初中高級程序員中，很多人曾問我如何從設(shè)計階段初期盡量避免日后帶來維護難度，今天就從Event Soucing（事件溯源）模式聊聊如何優(yōu)化我們的業(yè)務(wù)系統(tǒng)。

枯燥的理論知識避不可免，下面我盡量以代碼形式去演示事件驅(qū)動給在我們業(yè)務(wù)編程中帶來的好處。

什么是Event Sourcing ?

簡單來說，大家應(yīng)該都知道m(xù)ysql 數(shù)據(jù)同步中的binlog模式，我們在執(zhí)行一條查詢語句 select * from Person limit 1 時看到的數(shù)據(jù)可以理解為當(dāng)前時間的快照，眼前的這條數(shù)據(jù)，可能經(jīng)歷過若干update語句之后才得到的結(jié)果，事件溯源亦如此，如果我們把某一行數(shù)據(jù) 看做Person對象，一個對象從開始到消亡會經(jīng)歷過很多事件（update語句），如果我們要還原某個時間點的對象，只需依據(jù)的產(chǎn)生日期，按照順序在初始化對象上依次疊加上去，就能還原這一時期的對象了,

舉個例子一個person（張三）對象

Person zs = new Person(); 張三出生了

6歲 ?? 學(xué)生

25歲 ?? 警察

60歲 ?? 退休老人

雖然都是張三對象，但是不同時間段里張三的身份截然不同，如果我們要獲取警察時代的zs，我們用初始得到的zs依次累加上學(xué)生時代，警察時代就可以得到這一時代的zs對象了。

由此來看，對象好像顯得已經(jīng)不那么重要，事件溯源更加具有意義，因為它完整描述了這個對象從出生到消亡的全過程，也可以看為不斷在改變對象的狀態(tài)，事件是只會增加不會修改，對于現(xiàn)如今大數(shù)據(jù)時代，事件的產(chǎn)生對于數(shù)據(jù)挖掘、數(shù)據(jù)分析更具有意義。

扯了這么多，還是要以代碼來實際說說事件驅(qū)動帶來的好處，先看一處經(jīng)典的代碼

StockService.java

@Service
@AllArgsConstructor
public class StockService extends BaseMapper<Product> {
    //京東服務(wù)
    private final JdService jdService;
    //淘寶服務(wù)
    private final TaobaoService productService;
    //有贊服務(wù)
    private final YouzanService youzanService;
    //拼多多服務(wù)
    private final pddService pddService;
    //更多服務(wù)
    ...
    //設(shè)置商品庫存
    @Override
    public void changeProductStock(ChangeProductStockInputDTO inputDTO) {
        if(inputDTO.getStock<0){
          throw new BusinessException("庫存不能小于0");
        }
        Product product = baseMapper.getById(inputDTO.getId());
        product.setStock(inputDTO.getStock());
        baseMapper.updateById(product);
        //通知京東
        jdService.notify();
        //通知淘寶
        productService.notify();
        //通知有贊
        youzanService.notify();
        //更多需要執(zhí)行的業(yè)務(wù)...
    }
}

Product.java

@Data
public class Product {
    //id
    private String id;
    //庫存
    private BigDecimal stock;
    //...
}

例如比如在電商系統(tǒng)中，在我們自己的商品后臺中修改商品庫存后，我們要依次告知在其他第三方平臺這個商品庫存信息，我相信很多同學(xué)都會這樣寫的吧，這樣的代碼確實可以完成我們的業(yè)務(wù)功能，但隨著業(yè)務(wù)功能的復(fù)雜度提升，加上我們面向過程的編碼模式，一定會越加復(fù)雜，曾看到有將近5000多行的一個訂單類，相信不管誰看見這樣的類都會頭大，接下來我們就要想辦法優(yōu)化它，安排！

首先存在這樣的代碼是因為沒有劃清邊界，沒有保持一個領(lǐng)域中的純粹性，從StockService中注入大量的服務(wù)類與標(biāo)志性的貧血模型Product對象就能看出，既然我們提倡以高內(nèi)聚低耦合去編寫代碼，那首先去修改我們的Product吧，讓它變得豐富起來。

改變的Product.java

@Data
public class Product {
    public void changeStock(BigDecimal stock){
      if(delStatus == 1){
        throw new BusinessException("商品信息不存在");
      }
      if(stock < 0){
        throw new BusinessException("庫存不能小于0");
      }
      this.stock = stock;
      EventBus.instance().register(new ChangedProductStockDomainEvent(this));
    }
    //id
    private String id;
    //庫存
    private BigDecimal stock;
    //刪除狀態(tài)
    private int delStatus;
    //...
}
//名字盡量起得生動一些，單詞語法的過去式，現(xiàn)在進行時都具有意義
@Getter
@AllArgsConstructor
public class ChangedProductStockDomainEvent {
   private Product product;
}

更改的 StockService.java

@Service
@AllArgsConstructor
public class StockService extends BaseMapper<Product> {
    //設(shè)置商品庫存
    @Override
    public void changeProductStock(ChangeProductStockInputDTO inputDTO) {
        Product product = baseMapper.getById(inputDTO.getId());
        product.setProductStock(inputDTO.getProductStock());
    }
}

更改過后的代碼是不是看起來清爽了很多，加上我們賦予了Product對象方法之后，職責(zé)看起來就更加明確，充血模型體現(xiàn)出聚合內(nèi)單一的行為，在Product中我們只描述了此領(lǐng)域范圍的職能，已經(jīng)充分體現(xiàn)了高內(nèi)聚低耦合的思想，不參合其他業(yè)務(wù)邏輯。這時可能有的同學(xué)會問那怎么持久化到數(shù)據(jù)庫呢？在我工作的這些年里，遇到很多程序員，不論初中高級程序員都習(xí)慣了先建立數(shù)據(jù)庫，再去建立模型，但是我們要改變傳統(tǒng)思維，我們寫代碼是面向?qū)ο?，面向?qū)ο?，面向?qū)ο螅ㄖ匾氖虑檎f三遍），不是面向數(shù)據(jù)或者過程，在剝離了數(shù)據(jù)后，其實我們真正就做到了數(shù)據(jù)與業(yè)務(wù)代碼的剝離，下面我在說這樣具體的好處。

細(xì)心的同學(xué)看到我在Product的changeStock方法里，在執(zhí)行完一些邏輯判斷后，設(shè)置完商品庫存后，我們在EventBus 事件總線中注冊了一個事件，這個事件還沒有具體的作用，我們看看EventBus的實現(xiàn)

StockService.java

public class EventBus {
    public static EventBus instance() {
        return new EventBus();
    }
    private static final ThreadLocal<List<DomainEvent>> domainEvents = new ThreadLocal<>();
    public void init() {
        if (domainEvents.get() == null) {
            domainEvents.set(new ArrayList<>());
        }
    }
    public EventBus register(DomainEvent domainEvent) {
        List<DomainEvent> domainEventList = domainEvents.get();
        if (domainEventList == null)
            throw new IllegalArgumentException("domainEventList not init");
        domainEventList.add(domainEvent);
        return this;
    }
    /**
     * 獲取領(lǐng)域事件
     *
     * @return
     */
    public List<DomainEvent> getDomainEvent() {
        return domainEvents.get();
    }
    /**
     * 請空領(lǐng)域事件集合
     */
    public void reset() {
        domainEvents.set(null);
    }
}

在當(dāng)前線程內(nèi)內(nèi)存空間我們吧事件塞了進去，目前只有存儲作用，接下來我們要定義它的處理者

DomainEventProcessor.java

@Aspect
@Component
@Slf4j
public class DomainEventProcessor {
    /**
     * 這里我是我對RocketMq的封裝
     */
    @Autowired
    private EventPublisherExecutor processPublisherExecutor;
    /**
     * 當(dāng)前上下文內(nèi)訂閱者
     */
    @Autowired
    protected ApplicationContext applicationContext;
    private static ThreadLocal<AtomicInteger> counter = new ThreadLocal<>();
    @Pointcut("within(com.github.tom.*.application..*)")
    public void aopRule() {
    }
    /**
     * 為當(dāng)前線程初始化EventBus
     */
    @Before("aopRule()")
    public void initEventBus(JoinPoint joinPoint) {
        log.debug("初始化領(lǐng)域事件槽");
        log.debug("切入切點信息：" + joinPoint.getSignature().toString());
        EventBus.instance().init();
        if (counter.get() == null) {
            counter.set(new AtomicInteger(0));
        }
        counter.get().incrementAndGet();
    }
    /**
     * 發(fā)布領(lǐng)域事件
     */
    @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) {
                    //進程內(nèi)事件
                    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();
        //釋放計數(shù)器
        counter.set(null);
    }
}

這里借助了AOP功能，在AOP內(nèi)我對service進行攔截，在執(zhí)行方法攔截的出口時，查找當(dāng)前線程內(nèi)的EventBus中看是否有存在的領(lǐng)域事件，接下來把事件發(fā)送出去，事件的響應(yīng)分為進程內(nèi)和進程外（多微服務(wù)），剛才的同學(xué)問的如何持久化到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("數(shù)據(jù)操作錯誤");
        }
    }
}

在響應(yīng)事件的其中一個訂閱者，可以完成數(shù)據(jù)庫的持久化操作。接下來我們?nèi)ザx各個響應(yīng)ChangedProductStockDomainEvent事件的訂閱者就行，例如京東服務(wù)

@Component
public class JdStockEventHandler {
    @Autowired
    private JdAppService jdAppService;
    /**
     * 庫存持久化事件
     *
     * @param eventData
     */
    @StreamListener(value = "product-channel")
    public void receive(@Payload ChangedProductStockDomainEvent eventData) {
        jdAppService.changingInventory(eventData);
    }
}

事件驅(qū)動的模型大大降低了業(yè)務(wù)模塊耦合嚴(yán)重，在每個聚合的領(lǐng)域內(nèi)，我們應(yīng)該著重自身聚合的業(yè)務(wù)邏輯，事件的消費我們可以通過廣播通知和最終一致性來達成目的。業(yè)務(wù)代碼的純粹，也更適合TDD只對業(yè)務(wù)編寫測試代碼，例如我在編寫設(shè)置庫存的測試方法時，我只要構(gòu)造好商品對象，就可以按照測試用例編寫不同情況下的測試代碼了。

@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");
    }
}

好了今天的介紹就先這么多，后面我會介紹如何讓三層架構(gòu)中的Service層升級，變得充滿業(yè)務(wù)味道（領(lǐng)域服務(wù)）。

以上就是Event Sourcing事件溯源模式優(yōu)化業(yè)務(wù)系統(tǒng)的詳細(xì)內(nèi)容，更多關(guān)于Event Sourcing事件溯源的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: