Redmi Buds 4開箱簡評：才199還有降噪 可以無腦入
在上個月舉辦的Redmi Note11T Pro系列新機發布會上，除了兩款手機新品之外，Redmi還帶來了兩款TWS真無線藍牙耳機產品，Redmi Buds 4和Redmi Buds 4 Pro，此前我們在Redmi Note11T
圖片
為了保證數據一致性，ByteHouse提供了以下能力：
保障數據一致性最簡單的方式是通過狀態機來監聽流程執行過程：
圖片
在流程監控的架構設計中，主要包含以下：
圖片
圖片
流程引擎的核心為“責任鏈”，按照責任鏈上的節點順序依次執行所有任務，所以我們需要的三個基本單元分別為：
在研發工作中，我們時常會遇到以下問題：
最簡單的處理辦法，是讓node使用相同的上下文信息，將整個執行node模版化。我們讓所有的執行節點node實現相同的接口Delegation，統一使用相同的上下文executionContext作為執行方法的入參。
對于流程中的request和response，我們可以放入executionContext中，讓每個執行節點都可以通過上下文操作response。
// Delegation -type Delegation interface { Execute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError TryExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError ConfirmExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError CancelExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError Code() string Type() value.DelegationType}如果確定好了最小的執行節點，我們需要考慮到，業務場景并不會永遠順序執行node，再返回結果，流程執行過程中跳轉、循環、并發執行都是比較常見的操作?？紤]不同業務場景復用性，我們在執行節點之上加了一層執行策略，用策略behaivor來重新編排觸發執行節點的任務。
圖片
// ActivityBehavior -type ActivityBehavior interface { Enter(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppError Execute(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppError Leave(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppError Code() value.ActivityBehaviorCode}策略behavior提供有Enter，Execute，Leave三個接口，Enter負責生成執行節點任務instance，Execute負責編排并觸發執行任務instance操作，Leave負責跳轉到下一個behavior。
可以看出來策略behaivor的跳轉方式類似于鏈表，不斷執行next方法，所以編碼過程中需要注意不要出現死循環，小心stackoverflow。
執行器Executor的主要作用是串聯執行策略和執行節點，策略behavior將執行的命令下發給Executor，由Executor對執行節點的觸發操作。這里會根據執行節點的type，映射到三種執行節點的執行方式，包含tcc，執行一次，重試多次。
// DelegationExecutor -type DelegationExecutor interface { execute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError postExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError}func (de *DefaultDelegationExecutor) execute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppError { delegationCode := executionContext.GetExecutionInstance().GetDelegationCode() if len(delegationCode) == 0 || de.DelegationMap[delegationCode] == nil { logger.Info(ctx, "DefaultDelegationExecutor delegation code not found,use default delegation", zap.String("delegationCode", delegationCode)) delegationCode = string(value.DefaultDelegation) executionContext.GetExecutionInstance().SetDelegationCode(delegationCode) } return de.dumpExecute(ctx, executionContext, delegationCode)}func (de *DefaultDelegationExecutor) dumpExecute(ctx context.Context, executionContext ExecutionContextInterface, delegationCode string) apperror.AppError { FireEvent(ctx, executionContext, value.ExecutionStart) var err apperror.AppError delegation := de.DelegationMap[delegationCode] switch delegation.Type() { case value.TccDelegation: err = tccExecute(ctx, executionContext, delegation) case value.SingleDelegation: err = singleExecute(ctx, executionContext, delegation) case value.RetryDelegation: err = retryExecute(ctx, executionContext, delegation) } if err != nil { logger.Error(ctx, "delegation.Execute_err", zap.Error(err)) return apperror.Trace(err) } FireEvent(ctx, executionContext, value.ExecutionEnd) return nil}ExecutionContext上下文是用來記錄了流程執行的所有細節，包含以下：
為了保證整個流程執行的穩定性，這里除了response之外，所以其他的實例參數都不建議開放寫接口，response可以用來存儲流程實例執行過程中會產生的變量信息。
對于整個流程的定義ProcessEngineConfiguration，我們可以選擇最簡單的方式，即在數據庫里，將配置信息映射成json字符串。當然也可以選擇讀取配置文件，只要能滿足讀取方便，數據不丟即可。
// ExecutionContextInterface -type ExecutionContextInterface interface { GetProcessEngineConfiguration() ProcessEngineConfigurationInterface SetProcessEngineConfiguration(processEngineConfiguration ProcessEngineConfigurationInterface) GetExecutionInstance() instance.ExecutionInstanceInterface SetExecutionInstance(executionInstance instance.ExecutionInstanceInterface) GetActivityInstance() instance.ActivityInstanceInterface SetActivityInstance(activityInstance instance.ActivityInstanceInterface) GetProcessInstance() instance.ProcessInstanceInterface SetProcessInstance(processInstance instance.ProcessInstanceInterface) SetNeedPause(needPause bool) IsNeedPause() bool SetActivityIndex(activityIndex int) GetActivityIndex() int SetActivityBehaviorCode(activityBehaviorCode value.ActivityBehaviorCode) GetActivityBehaviorCode() value.ActivityBehaviorCode SetBizUniqueKey(bizUniqueKey string) GetBizUniqueKey() string GetRequest() map[string]interface{} SetRequest(request map[string]interface{}) GetResponse() map[string]string SetResponse(response map[string]string) AtomicAddResponse(key string, value string)}監聽器的主要作用是用來監聽流程執行中的重要參數信息。從上述executor接口可以看到fireEvent，它的作用是發送消息event，讓listener監聽到對應的event類型，完成一些定制化的行為。
類似于面向切面編程，我們可以在執行節點的前后增加定制化的邏輯，如打日志、監聽節點執行時間，持久化流程中產生的response信息、增加鏈路追蹤等。
圖片
最后，我們將上述的內容拼接串聯起來，主要提供三個接口：
process start(){ //1.get and create ProcessEngineConfigurationInterface 解析流程定義 //2.create processInstance 創建流程實例 //3.create ExecutionContext 創建執行上下文 //4. lockstrategy trylock //5. invoke process start processinstance.start() //6. persist processInstance and return //7. lockstrategy unlock }processinstance start(){ // get behavior // behavior enter behavior.Enter(ctx, executionContext) //behavior execute behavior.Execute(ctx, executionContext) //behavior leave behavior.Leave(ctx, executionContext)}相比于start，signal需要讀取執行的細節信息，找到之前失敗的執行節點位置，并加載到上下文中，再繼續執行。
對于失敗節點信息的持久化有兩種方式：第一，可以選擇在流程執行結束持久化；第二，可以通過listener在每個執行節點結束持久化。具體根據實際業務場景對于性能、數據一致性的要求做出抉擇。
輕量級流程引擎的基本功能到此已經實現，后續的擴展優化可以圍繞以下方向進行：
以下為簡單的processconfiguration的配置信息，此處使用DefaultBehavior，即同步順序執行策略。
{ "ProcessContentList":[ { "Behavior":"DefaultBehavior", "DelegationList":[ { "Code":"sample1" }， { "Code":"sample2" }, { "Code":"sample3" } ] }, { "Behavior":"DefaultBehavior", "DelegationList":[ { "Code":"sample4" }, { "Code":"sample5" } ] } ]}圖片
在listener里面加入日志，這樣可以追溯出整個流程的執行流程，以便更好的監控整個流程的運行狀態。
以ClickHouse集群縮容為例：
圖片
{ "ProcessContentList":[ // 查詢所有需要重分布的table { "Behavior":"DefaultBehavior",// 順序執行 "DelegationList":[ { "Code":"hor_reshard_table_loop" } ] }, // 遍歷所有table進行數據的重分布 { "LoopKey":"reshard_table_loop_key", "Behavior":"NonBlockLoopBehavior",// 非阻塞循環處理 "DelegationList":[ { "Code":"hor_reshard_table" } ] }, // 進行刪除節點操作 { "Behavior":"DefaultBehavior", "DelegationList":[ { "Code":"hor_start_remove_node" }, { "Code":"hor_prepare_node_vcloud", "PostCode":"hor_rollback_remove_node_vcloud"http:// 統一失敗回滾處理 }, { "Code":"hor_update_config_vcloud", "PostCode":"hor_rollback_remove_node_vcloud" }, { "Code":"hor_set_cluster_running", "PostCode":"hor_rollback_remove_node_vcloud" }, { "Code":"hor_release_node" }, { "Code":"hor_callback_bill" } ] } ]}一個流程引擎適配所有的業務場景幾乎是不可能，除非接受復雜的方案設計，而第三方流程引擎對于日常的業務開發顯得太笨重。輕量級流程引擎則會簡化接入方式，減少了過多http請求帶來的性能損耗，更加靈活多變，追述問題也變得簡單。
在ByteHouse中加入流程引擎的能力，能以較小的代價給業務更多重試的可能性，而不需要反復回滾，特別對于耗時很長的任務，能帶來更好用戶使用體驗。除此之外，流程引擎還能將業務流程模版化，增加接口服務的復用性，使得業務代碼的可讀性、擴展性得到提升，方便后期維護。
火山引擎云原生數據倉庫ByteHouse是火山引擎旗下的一款云原生數據倉庫，為用戶提供極速分析體驗，能夠支撐實時數據分析和海量數據離線分析，同時還具備便捷的彈性擴縮容能力，極致分析性能和豐富的企業級特性，助力客戶數字化轉型。
本文鏈接：http://m.www897cc.com/showinfo-26-10443-0.html火山引擎 ByteHouse：ClickHouse 如何保證海量數據一致性
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號