wordpress中文博客模板下载,烟台seo外包,国外招工信息最新招聘信息,上海平面设计师招聘背景 ClickHouse是一个开源的OLAP引擎#xff0c;不仅被全球开发者广泛使用#xff0c;在字节各个应用场景中也可以看到它的身影。基于高性能、分布式特点#xff0c;ClickHouse可以满足大规模数据的分析和查询需求#xff0c;因此字节研发团队以开源ClickHouse为基础… 背景 ClickHouse是一个开源的OLAP引擎不仅被全球开发者广泛使用在字节各个应用场景中也可以看到它的身影。基于高性能、分布式特点ClickHouse可以满足大规模数据的分析和查询需求因此字节研发团队以开源ClickHouse为基础推出火山引擎云原生数据仓库ByteHouse。 在日常工作中研发人员经常会遇到业务链路过长导致流程稳定性和数据一致性难保障的问题这在分布式、跨服务的场景中更为明显。本篇文章提出针对这一问题的解决思路在火山引擎ByteHouse中构建轻量级流程引擎来解决数据一致性问题。 使用轻量级流程引擎可以帮我们使用统一的标准来解决复杂业务链路的编排问题不仅提高业务代码的可读性和复用性还能更专注业务核心逻辑的开发让整体流程更加标准化、规范化。 总结来说使用流程引擎有以下优势 轻量级接入方便内存操作性能有保障易维护流程配置与业务分离支持热更新易扩展丰富的执行策略及算子支持 大体思路 上图为ByteHouse企业版管理平台功能架构图。从该功能架构图可以看出ByteHouse核心能力都是依赖ClickHouse集群对于集群节点多、数据计算量大的业务场景容易出现节点状态不一致的问题因此保证ClickHouse集群间的状态一致性是我们的核心诉求。 为了保证数据一致性ByteHouse提供了以下能力 event engine: 事件处理中心workflow engine轻量级流程引擎对账系统 保障数据一致性最简单的方式是通过状态机来监听流程执行过程 首先将所有的任务请求下发到event engine由event engine将任务分发对应的handler执行统一管理所有下发任务的生命周期并提供异步重试、回滚补偿等功能。流量汇总到event engine以后会让服务后续的业务扩展更加便捷。其次对于比较复杂的任务请求我们可以下发到workflow engine执行由workflow生成实例并编排任务队列管理流程执行实例的生命周期统一失败回滚失败重试。最后对于服务不可用等特殊场景产生的脏数据由对账服务兜底。 架构设计 在流程监控的架构设计中主要包含以下 流程管理层主要负责流程配置的解析初始化并完成编排策略的工作策略behavior层编排执行节点并下发执行任务到执行器执行器管理执行节点执行执行节点负责业务具体实现 实现方案 执行节点 流程引擎的核心为“责任链”按照责任链上的节点顺序依次执行所有任务所以我们需要的三个基本单元分别为 request入参processlist流程执行节点listresponse出参 在研发工作中我们时常会遇到以下问题 如果同时出现了一个问题node1、node2、node3之间的数据交互如何实现如果node1入参、出参与node2node3不一样该如何处理参数类型不同的node又该如何统一调度 最简单的处理办法是让node使用相同的上下文信息将整个执行node模版化。我们让所有的执行节点node实现相同的接口Delegation统一使用相同的上下文executionContext作为执行方法的入参。 对于流程中的request和response我们可以放入executionContext中让每个执行节点都可以通过上下文操作response。 // Delegation -
type Delegation interface {Execute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppErrorTryExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppErrorConfirmExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppErrorCancelExecute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppErrorCode() stringType() value.DelegationType
} 执行策略 如果确定好了最小的执行节点我们需要考虑到业务场景并不会永远顺序执行node再返回结果流程执行过程中跳转、循环、并发执行都是比较常见的操作。考虑不同业务场景复用性我们在执行节点之上加了一层执行策略用策略behaivor来重新编排触发执行节点的任务。 下图将流程分成了behavior1和behavior2分别对应不同的策略。简单的策略举例按顺序执行、并发执行、循环执行、条件跳转执行等。我们可以根据自身业务实际需要定制后续会有实例介绍。 // ActivityBehavior -
type ActivityBehavior interface {Enter(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppErrorExecute(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppErrorLeave(ctx context.Context, executionContext ExecutionContextInterface, pvmActivity PvmActivity) apperror.AppErrorCode() value.ActivityBehaviorCode
} 策略behavior提供有EnterExecuteLeave三个接口Enter负责生成执行节点任务instanceExecute负责编排并触发执行任务instance操作Leave负责跳转到下一个behavior。 可以看出来策略behaivor的跳转方式类似于链表不断执行next方法所以编码过程中需要注意不要出现死循环小心stackoverflow。 Executor 执行器Executor的主要作用是串联执行策略和执行节点策略behavior将执行的命令下发给Executor由Executor对执行节点的触发操作。这里会根据执行节点的type映射到三种执行节点的执行方式包含tcc执行一次重试多次。 // DelegationExecutor -
type DelegationExecutor interface {execute(ctx context.Context, executionContext ExecutionContextInterface) apperror.AppErrorpostExecute(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.AppErrordelegation : 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 ExecutionContext上下文是用来记录了流程执行的所有细节包含以下 ProcessEngineConfigurationInterface: 流程定义信息ExecutionInstanceInterface: 执行节点实例ActivityInstanceInterface: 执行策略实例ProcessInstanceInterface: 流程实例request入参response返回值 为了保证整个流程执行的稳定性这里除了response之外所以其他的实例参数都不建议开放写接口response可以用来存储流程实例执行过程中会产生的变量信息。 对于整个流程的定义ProcessEngineConfiguration我们可以选择最简单的方式即在数据库里将配置信息映射成json字符串。当然也可以选择读取配置文件只要能满足读取方便数据不丢即可。 // ExecutionContextInterface -
type ExecutionContextInterface interface {GetProcessEngineConfiguration() ProcessEngineConfigurationInterfaceSetProcessEngineConfiguration(processEngineConfiguration ProcessEngineConfigurationInterface)GetExecutionInstance() instance.ExecutionInstanceInterfaceSetExecutionInstance(executionInstance instance.ExecutionInstanceInterface)GetActivityInstance() instance.ActivityInstanceInterfaceSetActivityInstance(activityInstance instance.ActivityInstanceInterface)GetProcessInstance() instance.ProcessInstanceInterfaceSetProcessInstance(processInstance instance.ProcessInstanceInterface)SetNeedPause(needPause bool)IsNeedPause() boolSetActivityIndex(activityIndex int)GetActivityIndex() intSetActivityBehaviorCode(activityBehaviorCode value.ActivityBehaviorCode)GetActivityBehaviorCode() value.ActivityBehaviorCodeSetBizUniqueKey(bizUniqueKey string)GetBizUniqueKey() stringGetRequest() map[string]interface{}SetRequest(request map[string]interface{})GetResponse() map[string]stringSetResponse(response map[string]string)AtomicAddResponse(key string, value string)
} Listener 监听器的主要作用是用来监听流程执行中的重要参数信息。从上述executor接口可以看到fireEvent它的作用是发送消息event让listener监听到对应的event类型完成一些定制化的行为。 类似于面向切面编程我们可以在执行节点的前后增加定制化的逻辑如打日志、监听节点执行时间持久化流程中产生的response信息、增加链路追踪等。 API 最后我们将上述的内容拼接串联起来主要提供三个接口 Start: 启动流程Signal: 暂停或是异常退出后继续执行流程Abort: 强制中断流程 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 enterbehavior.Enter(ctx, executionContext)//behavior executebehavior.Execute(ctx, executionContext)//behavior leavebehavior.Leave(ctx, executionContext)
} 相比于startsignal需要读取执行的细节信息找到之前失败的执行节点位置并加载到上下文中再继续执行。 对于失败节点信息的持久化有两种方式第一可以选择在流程执行结束持久化第二可以通过listener在每个执行节点结束持久化。具体根据实际业务场景对于性能、数据一致性的要求做出抉择。 并发场景考虑 behavior策略中肯定会出现定制、并发、处理多个执行节点到场景的问题如果同时修改必定会造成数据错乱。简单的方法推荐使用带锁的容器存储可以被修改的信息(response)此处使用的是github.com/bytedance/gopkg包里面封装的skipmap。lockstrategy可以自己定义最适配业务场景的最简单的方案是redis锁同时也考虑到系统异常退出后的恢复问题。可以参考redis官网解决特殊情况下的锁异常解决方案https://redis.io/commands/setnx/ 后续的工作 轻量级流程引擎的基本功能到此已经实现后续的扩展优化可以围绕以下方向进行 界面化展示可以将链路执行情况展示出来策略behavior维度扩展适配各种业务场景增加子流程的维度可以复用原先的执行逻辑 Demo示例 以下为简单的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// 统一失败回滚处理},{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://www.morning.bzkgn.cn.gov.cn.bzkgn.cn http://www.morning.fosfox.com.gov.cn.fosfox.com http://www.morning.tgqzp.cn.gov.cn.tgqzp.cn http://www.morning.nj-ruike.cn.gov.cn.nj-ruike.cn http://www.morning.pwqyd.cn.gov.cn.pwqyd.cn http://www.morning.wckrl.cn.gov.cn.wckrl.cn http://www.morning.hmjasw.com.gov.cn.hmjasw.com http://www.morning.qlxgc.cn.gov.cn.qlxgc.cn http://www.morning.rqkck.cn.gov.cn.rqkck.cn http://www.morning.xhklb.cn.gov.cn.xhklb.cn http://www.morning.fhntj.cn.gov.cn.fhntj.cn http://www.morning.sqxr.cn.gov.cn.sqxr.cn http://www.morning.mxftp.com.gov.cn.mxftp.com http://www.morning.kkjlz.cn.gov.cn.kkjlz.cn http://www.morning.rdlong.com.gov.cn.rdlong.com http://www.morning.xsjfk.cn.gov.cn.xsjfk.cn http://www.morning.cprls.cn.gov.cn.cprls.cn http://www.morning.jglqn.cn.gov.cn.jglqn.cn http://www.morning.nhzxd.cn.gov.cn.nhzxd.cn http://www.morning.qhmhz.cn.gov.cn.qhmhz.cn http://www.morning.fplqh.cn.gov.cn.fplqh.cn http://www.morning.cpktd.cn.gov.cn.cpktd.cn http://www.morning.hlfgm.cn.gov.cn.hlfgm.cn http://www.morning.wnqbf.cn.gov.cn.wnqbf.cn http://www.morning.nwfpl.cn.gov.cn.nwfpl.cn http://www.morning.gthc.cn.gov.cn.gthc.cn http://www.morning.bpmtg.cn.gov.cn.bpmtg.cn http://www.morning.yfcbf.cn.gov.cn.yfcbf.cn http://www.morning.krqhw.cn.gov.cn.krqhw.cn http://www.morning.ghlyy.cn.gov.cn.ghlyy.cn http://www.morning.gbsfs.com.gov.cn.gbsfs.com http://www.morning.ftnhr.cn.gov.cn.ftnhr.cn http://www.morning.jglqn.cn.gov.cn.jglqn.cn http://www.morning.mbaiwan.com.gov.cn.mbaiwan.com http://www.morning.fktlr.cn.gov.cn.fktlr.cn http://www.morning.spdyl.cn.gov.cn.spdyl.cn http://www.morning.qhydkj.com.gov.cn.qhydkj.com http://www.morning.dfffm.cn.gov.cn.dfffm.cn http://www.morning.zwgbz.cn.gov.cn.zwgbz.cn http://www.morning.pjtnk.cn.gov.cn.pjtnk.cn http://www.morning.mkkcr.cn.gov.cn.mkkcr.cn http://www.morning.lrplh.cn.gov.cn.lrplh.cn http://www.morning.mlnbd.cn.gov.cn.mlnbd.cn http://www.morning.bypfj.cn.gov.cn.bypfj.cn http://www.morning.skkmz.cn.gov.cn.skkmz.cn http://www.morning.wffxr.cn.gov.cn.wffxr.cn http://www.morning.lqlhw.cn.gov.cn.lqlhw.cn http://www.morning.ywtbk.cn.gov.cn.ywtbk.cn http://www.morning.mxtjl.cn.gov.cn.mxtjl.cn http://www.morning.rxkq.cn.gov.cn.rxkq.cn http://www.morning.btgxf.cn.gov.cn.btgxf.cn http://www.morning.zfqr.cn.gov.cn.zfqr.cn http://www.morning.nzcgj.cn.gov.cn.nzcgj.cn http://www.morning.ttryd.cn.gov.cn.ttryd.cn http://www.morning.dwzwm.cn.gov.cn.dwzwm.cn http://www.morning.kdtdh.cn.gov.cn.kdtdh.cn http://www.morning.hwljx.cn.gov.cn.hwljx.cn http://www.morning.tpqrc.cn.gov.cn.tpqrc.cn http://www.morning.aswev.com.gov.cn.aswev.com http://www.morning.pwmpn.cn.gov.cn.pwmpn.cn http://www.morning.jmllh.cn.gov.cn.jmllh.cn http://www.morning.leeong.com.gov.cn.leeong.com http://www.morning.lpqgq.cn.gov.cn.lpqgq.cn http://www.morning.jgncd.cn.gov.cn.jgncd.cn http://www.morning.pbbzn.cn.gov.cn.pbbzn.cn http://www.morning.hptbp.cn.gov.cn.hptbp.cn http://www.morning.snyqb.cn.gov.cn.snyqb.cn http://www.morning.yhplt.cn.gov.cn.yhplt.cn http://www.morning.gxtbn.cn.gov.cn.gxtbn.cn http://www.morning.dbphz.cn.gov.cn.dbphz.cn http://www.morning.tsycr.cn.gov.cn.tsycr.cn http://www.morning.lpzqd.cn.gov.cn.lpzqd.cn http://www.morning.fssmx.com.gov.cn.fssmx.com http://www.morning.tnfyj.cn.gov.cn.tnfyj.cn http://www.morning.yltyr.cn.gov.cn.yltyr.cn http://www.morning.pbmkh.cn.gov.cn.pbmkh.cn http://www.morning.c7630.cn.gov.cn.c7630.cn http://www.morning.cgdyx.cn.gov.cn.cgdyx.cn http://www.morning.ghrlx.cn.gov.cn.ghrlx.cn http://www.morning.ypktc.cn.gov.cn.ypktc.cn
