一个多模块系统的重构从10个独立服务到统一调度技术支持拓世网络技术开发部一、现状我们有一个系统里面拆了10个独立模块· 模块A管理运行环境· 模块B调度多个执行单元· 模块C编排工作流· 模块D自动调整参数· 模块E采集外部状态· 模块F存储数据关系· 模块G做逻辑判断· 模块H分布式数据同步· 模块I模拟预测· 模块J选择最优方案每个模块单独跑都没问题但合在一起就出问题了。问题1一个请求要串行调10个模块用户发起一个请求系统要依次调用A→B→C→D→E→F→G→H→I→J。每个模块平均耗时300ms加起来3秒以上。用户反馈太慢。问题2数据存了多份模块E存了一份状态模块F存了一份知识模块I又存了一份模型参数。改一个字段要同时改三个地方经常漏改导致数据不一致。问题3模块之间相互调用A调BB调CC调D……改一个模块影响十几个文件。上线前测试要测全链路成本很高。问题4各模块优化目标不同模块J追求转化率模块C追求响应速度两个方向冲突整体效果反而下降。二、解决方案不搞那么多独立模块用统一调度器来管理所有功能。调度器负责1. 维护当前要完成的目标列表按优先级排序2. 维护系统运行状态所有数据集中存一份3. 维护业务数据关系实体和关联4. 执行简单的规则判断5. 调整策略参数6. 分发执行任务结构变成这样统一调度器├── 目标队列决定先做什么├── 状态存储所有数据放一起├── 数据关系库实体关联├── 规则引擎if-else判断├── 参数调优自动调整配置└── 任务执行器调用外部功能六个组件只和调度器通信组件之间不直接调用。三、代码实现1. 目标队列pythonimport heapqclass GoalQueue:def __init__(self):self.goals {}self.queue []def add(self, desc, priority5):gid str(len(self.goals) 1)self.goals[gid] {id: gid, desc: desc, priority: priority, done: False}heapq.heappush(self.queue, (-priority, gid))return giddef get_next(self):while self.queue:_, gid self.queue[0]goal self.goals.get(gid)if goal and not goal[done]:return goalheapq.heappop(self.queue)return Nonedef finish(self, gid):if gid in self.goals:self.goals[gid][done] True2. 状态存储pythonclass StateStore:def __init__(self):self.data {}self.watchers {}def get(self, key, defaultNone):parts key.split(.)cur self.datafor p in parts:if isinstance(cur, dict) and p in cur:cur cur[p]else:return defaultreturn curdef set(self, key, value):parts key.split(.)cur self.datafor p in parts[:-1]:if p not in cur:cur[p] {}cur cur[p]old cur.get(parts[-1])cur[parts[-1]] valueself._notify(key, old, value)def snapshot(self):import copyreturn copy.deepcopy(self.data)def watch(self, pattern, callback):self.watchers.setdefault(pattern, []).append(callback)def _notify(self, key, old, new):for pattern, cbs in self.watchers.items():if self._match(key, pattern):for cb in cbs:try:cb(key, old, new)except:passdef _match(self, key, pattern):if pattern *:return Trueif pattern.endswith(*):return key.startswith(pattern[:-1])return key pattern3. 数据关系库pythonclass RelationStore:def __init__(self):self.entities {}self.links []self.cache {}def add_entity(self, eid, etype, **attrs):if eid not in self.entities:self.entities[eid] {_type: etype}self.entities[eid].update(attrs)self.cache.clear()def add_link(self, src, dst, ltype, propsNone):if src not in self.entities or dst not in self.entities:return Falseself.links.append({src: src, dst: dst, type: ltype, props: props or {}})self.cache.clear()return Truedef query(self, eidNone, ltypeNone):key f{eid}:{ltype}if key in self.cache:return self.cache[key]result {entities: [], links: []}if eid and eid in self.entities:result[entities].append(self.entities[eid])for link in self.links:if link[src] eid or link[dst] eid:result[links].append(link)if ltype:for link in self.links:if link[type] ltype:result[links].append(link)if link[src] in self.entities:result[entities].append(self.entities[link[src]])if link[dst] in self.entities:result[entities].append(self.entities[link[dst]])self.cache[key] resultreturn result4. 规则引擎pythonclass RuleEngine:def run(self, goal, state, relations):result {matches: [],actions: [],score: 0.0}# 传递关系如果A关联BB关联C则A关联Clinks relations.get(links, [])for l1 in links:for l2 in links:if l1[dst] l2[src]:result[matches].append({src: l1[src],dst: l2[dst],weight: 0.7})# 根据目标生成建议操作if goal:desc goal.get(desc, )if 提升 in desc or 增加 in desc:result[actions].append({name: analyze, priority: high})if 降低 in desc or 减少 in desc:result[actions].append({name: audit, priority: high})return result5. 参数调优pythonimport randomimport copyclass ParamOptimizer:def __init__(self, size10):self.size sizeself.pop []self.best Noneself.gen 0def init(self, templates):for t in templates:self.pop.append({params: copy.deepcopy(t), fitness: 0.0})while len(self.pop) self.size:self.pop.append({params: {actions: random.sample([pub, opt, mon], 2)},fitness: 0.0})def optimize(self, goal, state, rules_result):# 评估适应度for p in self.pop:p[fitness] self._evaluate(p, goal)self.pop.sort(keylambda x: x[fitness], reverseTrue)if not self.best or self.pop[0][fitness] self.best.get(fitness, 0):self.best copy.deepcopy(self.pop[0])# 生成下一代elite self.pop[:2]new_pop elite.copy()while len(new_pop) self.size:p1, p2 random.sample(elite, 2)child self._crossover(p1, p2)if random.random() 0.1:child self._mutate(child)new_pop.append(child)self.pop new_popself.gen 1return self.bestdef _evaluate(self, p, goal):score 0.0if goal:desc goal.get(desc, )params_str str(p[params])matches sum(1 for w in desc.split() if len(w) 2 and w in params_str)score min(matches / max(1, len(desc.split())), 1.0) * 0.5score random.random() * 0.5return scoredef _crossover(self, p1, p2):params {}for k in set(p1[params].keys()) | set(p2[params].keys()):if k in p1[params] and k in p2[params]:params[k] random.choice([p1[params][k], p2[params][k]])elif k in p1[params]:params[k] copy.deepcopy(p1[params][k])else:params[k] copy.deepcopy(p2[params][k])return {params: params, fitness: 0.0}def _mutate(self, p):mutated copy.deepcopy(p)params mutated[params]for k, v in params.items():if isinstance(v, (int, float)):params[k] v * random.uniform(0.8, 1.2)return mutated6. 任务执行器pythonimport asyncioimport aiohttpclass TaskExecutor:def __init__(self, workers3):self.queue asyncio.Queue()self.handlers {}self.workers workersself.running Falseself.tasks []self.session Noneself.stats {ok: 0, fail: 0}def register(self, name, func):self.handlers[name] funcasync def start(self):self.running Trueself.session aiohttp.ClientSession()for i in range(self.workers):t asyncio.create_task(self._worker(i))self.tasks.append(t)async def stop(self):self.running Falsefor t in self.tasks:t.cancel()await asyncio.gather(*self.tasks, return_exceptionsTrue)if self.session:await self.session.close()async def submit(self, actions):ids []for act in actions:tid str(len(ids) 1)await self.queue.put({id: tid,type: act.get(type),target: act.get(target),params: act.get(params, {})})ids.append(tid)return idsasync def _worker(self, wid):while self.running:try:task await self.queue.get()result await self._execute(task)if result.get(ok):self.stats[ok] 1else:self.stats[fail] 1self.queue.task_done()except asyncio.CancelledError:breakexcept Exception as e:print(fworker {wid} error: {e})async def _execute(self, task):ttype task.get(type)target task.get(target)params task.get(params, {})try:if ttype handler:if target not in self.handlers:return {ok: False, err: fhandler not found: {target}}func self.handlers[target]if asyncio.iscoroutinefunction(func):result await func(**params)else:result func(**params)return {ok: True, result: result}elif ttype api:return await self._call_api(target, params)else:return {ok: False, err: funknown type: {ttype}}except Exception as e:return {ok: False, err: str(e)}async def _call_api(self, url, params):method params.get(method, GET)timeout aiohttp.ClientTimeout(totalparams.get(timeout, 30))async with self.session.request(methodmethod,urlurl,jsonparams.get(json),paramsparams.get(params),timeouttimeout) as resp:data await resp.json()return {ok: resp.status 400, status: resp.status, data: data}四、组装调度器pythonclass Scheduler:def __init__(self):self.goals GoalQueue()self.state StateStore()self.relations RelationStore()self.rules RuleEngine()self.optimizer ParamOptimizer()self.executor TaskExecutor()self.running Falseself.loop_task Noneasync def start(self):self.running Trueawait self.executor.start()self.loop_task asyncio.create_task(self._main_loop())print(调度器已启动)async def stop(self):self.running Falseif self.loop_task:self.loop_task.cancel()await self.executor.stop()print(调度器已停止)async def _main_loop(self):while self.running:try:goal self.goals.get_next()if not goal:await asyncio.sleep(1)continuestate self.state.snapshot()rels self.relations.query()rules_result self.rules.run(goal, state, rels)best self.optimizer.optimize(goal, state, rules_result)actions best.get(params, {}).get(actions, [])if actions:task_list [{type: handler, target: a, params: {}} for a in actions]ids await self.executor.submit(task_list)print(f执行任务: {ids})except asyncio.CancelledError:breakexcept Exception as e:print(f主循环异常: {e})await asyncio.sleep(2)五、使用示例假设我们有三个功能函数pythondef generate_content(topicAI):return f生成了关于{topic}的内容def publish_content(content):return f发布了: {content[:20]}...def check_rank(keywordAI):return {rank: 5, keyword: keyword}启动系统pythonimport asyncioasync def main():scheduler Scheduler()# 注册功能scheduler.executor.register(generate, generate_content)scheduler.executor.register(publish, publish_content)scheduler.executor.register(check, check_rank)# 设置目标scheduler.goals.add(提升SEO排名, 10)scheduler.goals.add(持续发布内容, 8)# 初始化参数模板templates [{actions: [generate, publish, check]},{actions: [generate, publish]},{actions: [publish, check]},]scheduler.optimizer.init(templates)# 启动await scheduler.start()# 运行30秒await asyncio.sleep(30)await scheduler.stop()print(f执行统计: {scheduler.executor.stats})if __name__ __main__:asyncio.run(main())六、效果对比在同一台服务器上测试指标 改之前10个模块直连 改之后统一调度平均响应时间 3.2秒 1.1秒每分钟处理量 30个 85个模块间调用开销 38% 12%改一个模块影响文件数 10 2-3七、总结这次重构的核心就两点1. 统一调度所有模块只和调度器通信不直接相互调用2. 数据集中状态、关系、参数都放在调度器统一管理带来的好处· 响应时间从3秒降到1秒左右· 吞吐量翻了两倍多· 代码维护简单了改一个模块不用牵一发动全身
