豬齒魚平臺中的微服務資料一致性解決方案-有解無憂

眾所周知，微服務架構解決了很多問題，通過分解復雜的單體式應用，在功能不變的情況下，使應用被分解為多個可管理的服務，為采用單體式編碼方式很難實作的功能提供了模塊化的解決方案，同時，每個微服務獨立部署、獨立擴展，使得持續化集成成為可能，由此，單個服務很容易開發、理解和維護，

微服務架構為開發帶來了諸多好處的同時，也引發了很多問題，比如服務運維變得更復雜，服務之間的依賴關系更復雜，資料一致性難以保證，

本篇文章將討論和介紹豬齒魚是如何保障微服務架構的資料一致性的，

主要內容包括：

傳統應用使用本地事務保持一致性
多資料源下的分布式事務
微服務架構中應滿足資料最終一致性原則
使用Event Sourcing保證微服務的最終一致性
使用可靠事件模式保證微服務的最終一致性
使用Saga保證微服務的最終一致性

下面將通過一個實體來分別介紹這幾種模式，

在Choerodon 豬齒魚的 DevOps流程中，有這樣一個步驟，

用戶在Choerodon 平臺上創建一個專案；
DevOps 服務對應創建一個專案；
DevOps 為該專案在 Gitlab 上創建對應的group，

傳統應用使用本地事務保持一致性

在講微服務架構的資料一致性之前，先介紹一下傳統關系型資料庫是如何保證一致性的，從關系型資料庫中的ACID理論講起，

ACID 即資料庫事務正確執行的四個基本要素，分別是：

原子性（Atomicity）：要么全部完成，要么全部不完成，不存在中間狀態
一致性（Consistency）：事務必須始終保持系統處于一致的狀態
隔離性（Isolation）：事務之間相互隔離，同一時間僅有一個請求用于同一資料
持久性（Durability）：事務一旦提交，該事務對資料庫所作的更改便持久的保存在資料庫之中，并不會被回滾

可以通過使用資料庫自身的ACID Transactions，將上述步驟簡化為如下偽代碼：

transaction.strat();
createProject();
devopsCreateProject();
gitlabCreateGroup();
transaction.commit();

這個程序可以說是十分簡單，如果在這一程序中發生失敗，例如DevOps創建專案失敗，那么該事務做回滾操作，使得最終平臺創建專案失敗，由于傳統應用一般都會使用一個關系型資料庫，所以可以直接使用 ACID transactions，保證了資料本身不會出現不一致，為保證一致性只需要：開始一個事務，改變（插入，洗掉，更新）很多行，然后提交事務（如果有例外時回滾事務），

隨著業務量的不斷增長，單資料庫已經不足以支撐龐大的業務資料，此時就需要對應用和資料庫進行拆分，于此同時，也就出現了一個應用需要同時訪問兩個或者兩個以上的資料庫或多個應用分別訪問不同的資料庫的情況，資料庫的本地事務則不再適用，

為了解決這一問題，分布式事務應運而生，

多資料源下的分布式事務

想象一下，如果很多用戶同時對Choerodon 平臺進行創建專案的操作，應用接收的流量和業務資料劇增，一個資料庫并不足以存盤所有的業務資料，那么我們可以將應用拆分成IAM服務和DevOps服務，其中兩個服務分別使用各自的資料庫，這樣的情況下，我們就減輕了請求的壓力和資料庫訪問的壓力，兩個分別可以很明確的知道自己執行的事務是成功還是失敗，但是同時在這種情況下，每個服務都不知道另一個服務的狀態，因此，在上面的例子中，如果當DevOps創建專案失敗時，就無法直接使用資料庫的事務，

那么如果當一個事務要跨越多個分布式服務的時候，我們應該如何保證事務呢?

為了保證該事務可以滿足ACID，一般采用2PC或者3PC， 2PC（Two Phase Commitment Protocol），實作分布式事務的經典代表就是兩階段提交協議，2PC包括準備階段和提交階段，在此協議中，一個或多個資源管理器的活動均由一個稱為事務協調器的單獨軟體組件來控制，

我們為DevOps服務分配一個事務管理器，那么上面的程序可以整理為如下兩個階段：

準備階段：

提交/回滾階段：

2PC 提供了一套完整的分布式事務的解決方案，遵循事務嚴格的 ACID 特性，

但是，當在準備階段的時候，對應的業務資料會被鎖定，直到整個程序結束才會釋放鎖，如果在高并發和涉及業務模塊較多的情況下，會對資料庫的性能影響較大，而且隨著規模的增大，系統的可伸縮性越差，同時由于 2PC引入了事務管理器，如果事務管理器和執行的服務同時宕機，則會導致資料產生不一致，雖然又提出了3PC 將2PC中的準備階段再次一分為二的來解決這一問題，但是同樣可能會產生資料不一致的結果，

微服務架構中應滿足資料最終一致性原則

不可否認，2PC 和3PC 提供了解決分布式系統下事務一致性問題的思路，但是2PC同時又是一個非常耗時的復雜程序，會嚴重影響系統效率，在實踐中我們盡量避免使用它，所以在分布式系統下無法直接使用此方案來保證事務，

對于分布式的微服務架構而言，傳統資料庫的ACID原則可能并不適用，首先微服務架構自身的所有資料都是通過API 進行訪問，這種資料訪問方式使得微服務之間松耦合，并且彼此之間獨立非常容易進行性能擴展，其次不同服務通常使用不同的資料庫，甚至并不一定會使用同一類資料庫，反而使用非關系型資料庫，而大部分的非關系型資料庫都不支持2PC，

在這種情況下，又如何解決事務一致性問題呢？

一個最直接的辦法就是考慮資料的強一致性，根據Eric Brewer提出的CAP理論，只能在資料強一致性（C）和可用性（A）之間做平衡，

CAP 是指在一個分布式系統下，包含三個要素：Consistency（一致性）、Availability（可用性）、Partition tolerance（磁區容錯性），并且三者不可得兼，

一致性（Consistency），是指對于每一次讀操作，要么都能夠讀到最新寫入的資料，要么錯誤，所有資料變動都是同步的，
可用性（Availability），是指對于每一次請求，都能夠得到一個及時的、非錯的回應，但是不保證請求的結果是基于最新寫入的資料，即在可以接受的時間范圍內正確地回應用戶請求，
磁區容錯性（Partition tolerance），是指由于節點之間的網路問題，即使一些訊息丟包或者延遲，整個系統仍能夠提供滿足一致性和可用性的服務，

關系型資料庫單節點保證了資料強一致性（C）和可用性（A），但是卻無法保證磁區容錯性（P），

然而在分布式系統下，為了保證模塊的磁區容錯性（P），只能在資料強一致性（C）和可用性（A）之間做平衡，具體表現為在一定時間內，可能模塊之間資料是不一致的，但是通過自動或手動補償后能夠達到最終的一致，

可用性一般是更好的選擇，但是在服務和資料庫之間維護事務一致性是非常根本的需求，微服務架構中應該選擇滿足最終一致性，

那么我們應該如何實作資料的最終一致性呢？

使用Event Sourcing保證微服務的最終一致性

什么是Event Sourcing（事件溯源）？

一個物件從創建開始到消亡會經歷很多事件，傳統的方式是保存這個業務物件當前的狀態，但更多的時候，我們也許更關心這個業務物件是怎樣達到這一狀態的，Event Sourcing從根本上和傳統的資料存盤不同，它存盤的不是業務物件的狀態，而是有關該業務物件一系列的狀態變化的事件，只要一個物件的狀態發生變化，服務就需要自動發布事件來附加到事件的序列中，這個操作本質上是原子的，

現在將上面的訂單程序用Event Sourcing 進行改造，將訂單變動的一個個事件存盤起來，服務監聽事件，對訂單的狀態進行修改，

可以看到 Event Sourcing 完整的描述了物件的整個生命周期程序中所經歷的所有事件，由于事件是只會增加不會修改，這種特性使得領域模型十分的穩定，

Event sourcing 為整體架構提供了一些可能性，但是將應用程式的每個變動都封裝到事件保存下來，并不是每個人都能接受的風格，而且大多數人都認為這樣很別扭，同時這一架構在實際應用實踐中也不是特別的成熟，

使用可靠事件模式保證微服務的最終一致性

可靠事件模式屬于事件驅動架構，微服務完成操作后向訊息代理發布事件，關聯的微服務從訊息代理訂閱到該事件從而完成相應的業務操作，關鍵在于可靠事件投遞和避免事件重復消費，

可靠事件投遞有兩個特性：

每個服務原子性的完成業務操作和發布事件；
訊息代理確保事件投遞至少一次 (at least once)，避免重復消費要求消費事件的服務實作冪等性，

有兩種實作方式：

1. 本地事件表

本地事件表方法將事件和業務資料保存在同一個資料庫中，使用一個額外的“事件恢復”服務來恢復事件，由本地事務保證更新業務和發布事件的原子性，考慮到事件恢復可能會有一定的延時，服務在完成本地事務后可立即向訊息代理發布一個事件，

使用本地事件表將事件和業務資料保存在同一個資料庫中，會在每個服務存盤一份資料，在一定程度上會造成代碼的重復冗余，同時，這種模式下的業務系統和事件系統耦合比較緊密，額外增加的事件資料庫操作也會給資料庫帶來額外的壓力，可能成為瓶頸，

2. 外部事件表

針對本地事件表出現的問題，提出外部事件表方法，將事件持久化到外部的事件系統，事件系統需提供實時事件服務以接收微服務發布的事件，同時事件系統還需要提供事件恢復服務來確認和恢復事件，

借助Kafka和可靠事件，可以通過如下代碼實作訂單流程，

// IAM ProjectService

@Service
@RefreshScope
public class ProjectServiceImpl implements ProjectService {

    private ProjectRepository projectRepository;

    private UserRepository userRepository;

    private OrganizationRepository organizationRepository;

    @Value("${choerodon.devops.message:false}")
    private boolean devopsMessage;

    @Value("${spring.application.name:default}")
    private String serviceName;

    private EventProducerTemplate eventProducerTemplate;

    public ProjectServiceImpl(ProjectRepository projectRepository,
                              UserRepository userRepository,
                              OrganizationRepository organizationRepository,
                              EventProducerTemplate eventProducerTemplate) {
        this.projectRepository = projectRepository;
        this.userRepository = userRepository;
        this.organizationRepository = organizationRepository;
        this.eventProducerTemplate = eventProducerTemplate;
    }

    @Transactional(rollbackFor = CommonException.class)
    @Override
    public ProjectDTO update(ProjectDTO projectDTO) {
        ProjectDO project = ConvertHelper.convert(projectDTO, ProjectDO.class);
        if (devopsMessage) {
            ProjectDTO dto = new ProjectDTO();
            CustomUserDetails details = DetailsHelper.getUserDetails();
            UserE user = userRepository.selectByLoginName(details.getUsername());
            ProjectDO projectDO = projectRepository.selectByPrimaryKey(projectDTO.getId());
            OrganizationDO organizationDO = organizationRepository.selectByPrimaryKey(projectDO.getOrganizationId());
            ProjectEventPayload projectEventMsg = new ProjectEventPayload();
            projectEventMsg.setUserName(details.getUsername());
            projectEventMsg.setUserId(user.getId());
            if (organizationDO != null) {
                projectEventMsg.setOrganizationCode(organizationDO.getCode());
                projectEventMsg.setOrganizationName(organizationDO.getName());
            }
            projectEventMsg.setProjectId(projectDO.getId());
            projectEventMsg.setProjectCode(projectDO.getCode());
            Exception exception = eventProducerTemplate.execute("project", EVENT_TYPE_UPDATE_PROJECT,
                    serviceName, projectEventMsg, (String uuid) -> {
                        ProjectE projectE = projectRepository.updateSelective(project);
                        projectEventMsg.setProjectName(project.getName());
                        BeanUtils.copyProperties(projectE, dto);
                    });
            if (exception != null) {
                throw new CommonException(exception.getMessage());
            }
            return dto;
        } else {
            return ConvertHelper.convert(
                    projectRepository.updateSelective(project), ProjectDTO.class);
        }
    }
}

// DEVOPS DevopsEventHandler
@Component
public class DevopsEventHandler {

    private static final String DEVOPS_SERVICE = "devops-service";
    private static final String IAM_SERVICE = "iam-service";

    private static final Logger LOGGER = LoggerFactory.getLogger(DevopsEventHandler.class);

    @Autowired
    private ProjectService projectService;
    @Autowired
    private GitlabGroupService gitlabGroupService;

    private void loggerInfo(Object o) {
        LOGGER.info("data: {}", o);
    }

    /**
     * 創建專案事件
     */
    @EventListener(topic = IAM_SERVICE, businessType = "createProject")
    public void handleProjectCreateEvent(EventPayload<ProjectEvent> payload) {
        ProjectEvent projectEvent = payload.getData();
        loggerInfo(projectEvent);
        projectService.createProject(projectEvent);
    }

    /**
     * 創建組事件
     */
    @EventListener(topic = DEVOPS_SERVICE, businessType = "GitlabGroup")
    public void handleGitlabGroupEvent(EventPayload<GitlabGroupPayload> payload) {
        GitlabGroupPayload gitlabGroupPayload = payload.getData();
        loggerInfo(gitlabGroupPayload);
        gitlabGroupService.createGroup(gitlabGroupPayload);
    }
}

使用Saga保證微服務的最終一致性 - Choerodon的解決方案

Saga是來自于1987年Hector GM和Kenneth Salem論文，在他們的論文中提到，一個長活事務Long lived transactions (LLTs) 會相對較長的占用資料庫資源，如果將它分解成多個事務，只要保證這些事務都執行成功，或者通過補償的機制，來保證事務的正常執行，這一個個的事務被他們稱之為Saga，

Saga將一個跨服務的事務拆分成多個事務，每個子事務都需要定義一個對應的補償操作，通過異步的模式來完成整個Saga流程，

在Choerodon中，將專案創建流程拆分成多個Saga，

// ProjectService

    @Transactional
    @Override
    @Saga(code = PROJECT_CREATE, description = "iam創建專案", inputSchemaClass = ProjectEventPayload.class)
    public ProjectDTO createProject(ProjectDTO projectDTO) {

        if (projectDTO.getEnabled() == null) {
            projectDTO.setEnabled(true);
        }
        final ProjectE projectE = ConvertHelper.convert(projectDTO, ProjectE.class);
        ProjectDTO dto;
        if (devopsMessage) {
            dto = createProjectBySaga(projectE);
        } else {
            ProjectE newProjectE = projectRepository.create(projectE);
            initMemberRole(newProjectE);
            dto = ConvertHelper.convert(newProjectE, ProjectDTO.class);
        }
        return dto;
    }

    private ProjectDTO createProjectBySaga(final ProjectE projectE) {
        ProjectEventPayload projectEventMsg = new ProjectEventPayload();
        CustomUserDetails details = DetailsHelper.getUserDetails();
        projectEventMsg.setUserName(details.getUsername());
        projectEventMsg.setUserId(details.getUserId());
        ProjectE newProjectE = projectRepository.create(projectE);
        projectEventMsg.setRoleLabels(initMemberRole(newProjectE));
        projectEventMsg.setProjectId(newProjectE.getId());
        projectEventMsg.setProjectCode(newProjectE.getCode());
        projectEventMsg.setProjectName(newProjectE.getName());
        OrganizationDO organizationDO =
                organizationRepository.selectByPrimaryKey(newProjectE.getOrganizationId());
        projectEventMsg.setOrganizationCode(organizationDO.getCode());
        projectEventMsg.setOrganizationName(organizationDO.getName());
        try {
            String input = mapper.writeValueAsString(projectEventMsg);
            sagaClient.startSaga(PROJECT_CREATE, new StartInstanceDTO(input, "project", newProjectE.getId() + ""));
        } catch (Exception e) {
            throw new CommonException("error.organizationProjectService.createProject.event", e);
        }
        return ConvertHelper.convert(newProjectE, ProjectDTO.class);
    }

// DevopsSagaHandler
@Component
public class DevopsSagaHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(DevopsSagaHandler.class);
    private final Gson gson = new Gson();

    @Autowired
    private ProjectService projectService;
    @Autowired
    private GitlabGroupService gitlabGroupService;

    private void loggerInfo(Object o) {
        LOGGER.info("data: {}", o);
    }

    /**
     * 創建專案saga
     */
    @SagaTask(code = "devopsCreateProject",
            description = "devops創建專案",
            sagaCode = "iam-create-project",
            seq = 1)
    public String handleProjectCreateEvent(String msg) {
        ProjectEvent projectEvent = gson.fromJson(msg, ProjectEvent.class);
        loggerInfo(projectEvent);
        projectService.createProject(projectEvent);
        return msg;
    }

    /**
     * 創建組事件
     */
    @SagaTask(code = "devopsCreateGitLabGroup",
            description = "devops 創建 GitLab Group",
            sagaCode = "iam-create-project",
            seq = 2)
    public String handleGitlabGroupEvent(String msg) {
        ProjectEvent projectEvent = gson.fromJson(msg, ProjectEvent.class);
        GitlabGroupPayload gitlabGroupPayload = new GitlabGroupPayload();
        BeanUtils.copyProperties(projectEvent, gitlabGroupPayload);
        loggerInfo(gitlabGroupPayload);
        gitlabGroupService.createGroup(gitlabGroupPayload, "");
        return msg;
    }
}

可以發現，Saga和可靠事件模式很相似，都是將微服務下的事務作為一個個體，然后通過有序序列來執行，但是在實作上，有很大的區別，

可靠事件依賴于Kafka，消費者屬于被動監聽Kafka的訊息，鑒于Kafka自身的原因，如果對消費者進行橫向擴展，效果并不理想，

而在 Saga 中，我們為 Saga 分配了一個orchestrator作為事務管理器，當服務啟動時，將服務中所有的 SagaTask 注冊到管理器中，當一個 Saga 實體通過sagaClient.startSaga啟動時，服務消費者就可以通過輪詢的方式主動拉取到該實體對應的Saga資料，并執行對應的業務邏輯，執行的狀態可以通過事務管理器進行查看，展現在界面上，

通過Choerodon的事務定義界面，將不同服務的SagaTask 收集展示，可以看到系統中的所有Saga 定義以及所屬的微服務，同時，在每一個Saga 定義的詳情中，可以詳細的了解到該Saga的詳細資訊：

在這種情況下，當并發量增多或者 SagaTask 的數量很多的時候，可以很便捷的對消費者進行擴展，

Saga的補償機制

Saga支持向前和向后恢復：

向后恢復：如果任意一個子事務失敗，則補償所有已完成的事務
向前恢復：如果子事務失敗，則重試失敗的事務

Choerodon 采用的是向前恢復，通過界面可以很方便的對事務的資訊進行檢索，當Saga發生失敗時，也可以看到失敗的原因，并且手動進行重試，通過Choerodon的事務實體界面，可以查詢到系統中運行的所有Saga實體，掌握實體的運行狀態，并對失敗的實體進行手動的重試：

對于向前恢復而言，理論上我們的子事務最終總是會成功的，但是在實際的應用中，可能因為一些其他的因素，造成失敗，那么就需要有對應的故障恢復回滾的機制，

使用Saga的要求

Saga是一個簡單易行的方案，使用Saga的兩個要求：

冪等：冪等是每個Saga 多次執行所產生的影回應該和一次執行的影響相同，一個很簡單的例子，上述流程中，如果在創建專案的時候因為網路問題導致超時，這時如果進行重試，請求恢復，如果沒有冪等，就可能創建了兩個專案，
可交換：可交換是指在同一層級中，無論先執行那個Saga最終的結果都應該是一樣的，

綜合比較

2PC是一個阻塞式，嚴格滿足ACID原則的方案，但是因為性能上的原因在微服務架構下并不是最佳的方案，

Event sourcing 為整體架構提供了一些可能性，但是如果隨著業務的變更，事件結構自身發生一定的變化時，需要通過額外的方式來進行補償，而且當下并沒有一個成熟完善的框架，

基于事件驅動的可靠事件是建立在訊息佇列基礎上，每一個服務，除了自己的業務邏輯之外還需要額外的事件表來保存當前的事件的狀態，所以相當于是把集中式的事務狀態分布到了每一個服務當中，雖然服務之間去中心化，但是當服務增多，服務之間的分布式事務帶來的應用復雜度也再提高，當事件發生問題時，難以定位，

而Saga降低了資料一致性的復雜度，簡單易行，將所有的事務統一可視化管理，讓運維更加簡單，同時每一個消費者可以進行快速的擴展，實作了事務的高可用，

本文由豬齒魚技術團隊原創，轉載請注明出處

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/413004.html

標籤：其他

上一篇：深度介紹Flink在位元組跳動資料流的實踐

下一篇：影像處理技術（二）濾波去噪（下）