Zhuang's Diary

在大型国际银行的转型中，Core Banking system系统稳定但僵化，移动网银渠道却要求毫秒级响应与全天候在线。

架构设计 - 数据服务层

核心思想​：将散落在各个“烟囱式”系统（银行卡/信用卡支付、风控、保险等）的数据通过技术手段（CDC、日志解析）实时汇聚起来，构建统一、标准的数据模型，并通过 API 的方式透明地提供给前台应用。核心系统停机时，该数据层能提供​“Stand-In”应急服务——支持只读，甚至有限度的暂存式写入。

• 读写分离：读操作快速由数据服务层响应，写操作依赖主系统保障强一致性；
• 容灾能力提升：主系统故障时自动切换至 Stand-In 模式，关键服务仍可继续执行；
• 真正渐进替代：业务逻辑可逐步迁移至服务层，最终实现核心现代化，不需一蹴而就。

主要机制说明：

• 正常模式：
  • 所有读操作由数据服务层提供低延迟响应；
  • 写操作严格通过 Connector 等连接器进入Core Banking system系统，确保主记录一致性；
  • Core Banking system 通过 CDC（如 Debezium）捕获变更并发布至 Kafka，数据服务层实时消费、更新缓存/数据库，最终一致性得以保障。
• 容灾模式：
  • Core Banking system不可用时，API 网关自动切换路由；
  • 写操作由轻量 Stand-In 服务代为接收，校验后写入本地“待处理交易”队列，并立即反馈客户“处理中”状态；
  • 主系统恢复后，后台任务自动逐笔回放交易，完成结算和Core Banking system同步处理。

治理、风险与项目管控：核心现代化控制塔

借鉴 OliverWyman 建议的 Core Modernization Control Tower，可为该架构增设治理层，关键职责包括  ：

• 跨职能治理机制：聚焦业务、运营、技术多方 alignment，确保宏观战略与技术路径一致；
• 风险管理监督：对迁移过程中的操作风险、合规风险、系统风险进行持续审视；
• 迁移节奏策略：统筹“hollow-out”业务削减主机依赖的节奏与业务模块迁移优先级；
• 变更可视化与沟通机制：提供实时反馈与透明度，保障项目顺滑执行。

技术选型与数据架构策略

1. 流式同步与事件驱动

• Debezium + Kafka：主流国际银行采用 Debezium 捕获 DB2 或主机日志，实时推送事件至 Kafka；这已是金融业标准方案  。

2. 存储平台设计

• Redis 集群：用于缓存极热数据，支撑毫秒级读性能；
• TiDB / OceanBase：支持强一致性的分布式关系型存储；适用全量账户与交易数据场景。

3. 数据架构策略

• 结合 Data Mesh / Lakehouse 架构思路提升分析与共享能力；
• 服务层接口应遵循 BIAN / Information Framework 等银行业标准，实现语义统一与模块化服务边界；增强系统的互操作性与标准合规性。

我们得到了什么？

• 弹性与连续性的革命​：Stand-In模式将核心停机从“灾难”变为“可管理的服务降级”，极大提升了业务连续性。
• 解耦开发能力：产品团队可绕开主机耦合，直接基于数据服务层构建灾备能力，实时风控等功能，交付加快；
• 开放 API 驱动：支持开放银行、第三方金融机构接入，实现更灵活生态协同及合作创新，符合当前数字银行趋势。

Modern Core Banking and Wealth Platforms: Who’s Leading with Real Contracts (2019–2025)

https://www.avaloq.com/services/ecosystem

1. 概念

1. 幂等性 + 事务性
   • Kafka 的 enable.idempotence=true 解决了 生产端的消息重复发送问题，保证“至多一次” → “恰好一次”。
   • transactional.id + 事务 API 则让生产者能把多个写操作当作原子单元，要么全写入，要么全不写。
2. 消费者端隔离级别
   • isolation.level=read_committed 确保消费端只读取 提交成功的事务消息，不会读到失败/未提交的写入。
   • 这相当于 SQL 里的 READ COMMITTED 隔离级别。
3. 两阶段提交（2PC）与事务日志
   • Kafka 内部事务机制不是严格意义上的 分布式 XA 2PC，而是 Producer Coordinator + Transaction Log 的轻量实现。
   • 当你把多个 Topic/Partition 的消息放在一个事务里时，Kafka 会写一条 事务日志（txn log），由 Coordinator 来决定事务的提交/中止状态，消费者依赖 txn log 过滤未提交消息。
   • 所以它更像是 “Kafka 内部的两阶段提交”，如果要与外部数据库（如 MySQL、Redis）做一致性，仍需要 外部协调机制（典型是 Outbox Pattern）。

2. 工程实践

在金融支付类系统里，通常会结合以下手段来确保可靠性：

• 生产端配置最佳实践

1
2
3
4
5

enable.idempotence=true   # 解决了生产端的消息重复发送问题，保证“至多一次” → “恰好一次”。
acks=all    # 确保 ISR 同步确认，确保消息不重复
retries=Integer.MAX_VALUE    # 确保消息不丢失
max.in.flight=1   # 避免并发请求导致的乱序
transactional.id=tx-producer-001

• 消费端配置最佳实践

1
2
3
4

isolation.level=read_committed
enable.auto.commit=false   # 结合事务手动提交offset
# offset 也写入 Kafka 的事务，确保消息消费与结果提交原子化。
# 常见模式是 Kafka + DB 的事务性写入（即 Outbox Pattern）

• 幂等消费（Exactly Once Processing）
  • 即使 Kafka 层面保证了 “Exactly Once Delivery”，但在消费者更新外部存储（数据库、账本系统）时，仍要确保幂等性。
  • 常见做法：
    • 使用 唯一事务ID/流水号 作为幂等键
    • 在数据库侧维护 dedup 表/唯一索引，避免重复扣款

3. 金融场景案例

1. 支付宝/微信支付 - 支付链路
   • Kafka 事务常用于 支付事件流，比如：
     • 扣款事件（资金冻结）
     • 风控事件（交易审核）
     • 清算/记账事件（账本入账）
   • 如果任何一步失败，整个事务会被中止，消费者端不会看到“半成品交易”。
2. 证券交易所撮合系统
   • 撮合引擎把订单事件写入 Kafka，撮合成功后再写入交易结果事件，两者要么同时写入，要么同时回滚。
   • 避免出现 “订单被接收但未撮合成功” 这种不一致。
3. 银行账务系统（Outbox + Kafka EOS）
   • 常见设计是 双写一致性：
     • 业务库写账务流水（Outbox 表）
     • Kafka Connect CDC 监听 Outbox 表 → 推送到 Kafka → 消费到清算/风控服务
   • Outbox 表和账务写操作在同一数据库事务中完成，Kafka 端依赖 EOS 保证消息不丢不重。

总结：Kafka EOS 只能保证 Kafka 内部端到端不丢不重，如果要与外部系统（DB/账本）做事务一致性，需要使用 Outbox Pattern 或幂等消费 来兜底。

DBOS simplifies building reliable and observable programs. Add a few annotations to your code to make it resilient to any failure.

• https://github.com/dbos-inc
• https://www.dbos.dev/

1. business logic in normal code, with branches, loops, subtasks, and retries. DBOS makes it resilient to any failure.

1
2
3
4
5
6
7
8
9
10

@DBOS.step()
def validate_payment():
    ...

@DBOS.workflow()
def checkout_workflow()
    validate_payment()
    check_inventory()
    ship_order()
    notify_customer()

2. Launch any task to run in the background and guarantee it eventually completes.

1
2
3
4
5
6
7
8

@DBOS.workflow()
def schedule_reminder(to_email, days_to_wait):
    DBOS.recv(days_to_seconds(days_to_wait))
    send_reminder_email(to_email, days_to_wait)

@app.post("/email")
def email_endpoint(request):
    DBOS.start_workflow(schedule_reminder, request.email, request.days)

3. Schedule functions to run at specific times.

1
2
3
4
5
6

@DBOS.scheduled("0 * * * *") # Run once an hour
@DBOS.workflow()
def run_hourly(scheduled_time, actual_time):
    results = search_hackernews("serverless")
    for comment, url in results:
        post_to_slack(comment, url)

4. Build data pipelines that are reliable and observable by default. DBOS durable queues guarantee all your tasks complete.

1
2
3
4
5
6
7
8

queue = Queue("indexing_queue")

@DBOS.workflow()
def indexing_workflow(urls):
    handles = []
    for url in urls:
        handles.append(queue.enqueue(index_step, url))
    return [h.get_result() for h in handles]

5. Consume Kafka messages exactly-once, no need to worry about timeouts or offsets.

1
2
3
4
5
6

@DBOS.kafka_consumer(config,["alerts-topic"])
@DBOS.workflow()
def process_kafka_alerts(msg):
    alerts = msg.value.decode()
    for alert in alerts:
        respond_to_alert(alert)

https://docs.dbos.dev/python/tutorials/kafka-integration

6. Use durable workflows to build reliable, fault-tolerant AI agents. Integrate with popular frameworks like LangChain and LlamaIndex.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

@DBOS.workflow()
def agentic_research_workflow(topic, max_iterations):
  research_results = []
  for i in range(max_iterations):
    research_result = research_query(topic)
    research_results.append(research_result)
    if not should_continue(research_results):
      break
    topic = generate_next_topic(topic, research_results)
  return synthesize_research_report(research_results)

@DBOS.step()
def research_query(topic):
  ...

在金融系统中的优势分析：

1. 天然适合银行交易场景：DBOS的工作流模式完美匹配银行的多步骤交易处理需求
2. PostgreSQL基础：许多银行已经在使用PostgreSQL，技术栈融合度高
3. 故障恢复能力：关键业务操作的自动恢复功能对银行至关重要
4. 代码简洁性：通过注解方式实现持久化，降低开发复杂度

潜在挑战：

1. 监管合规：需要验证DBOS是否满足银行业的严格审计要求
2. 性能要求：大型银行的高并发交易处理能力需要充分测试
3. 企业集成：与现有核心银行系统的集成复杂性
4. 安全标准：是否符合PCI-DSS、SOX等金融行业安全标准

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// case 1：客户服务流程
@DBOS.workflow()
async function loanApplication(applicationId: string) {
  await documentVerification(applicationId);
  await creditScoreCheck(applicationId);  
  await riskAssessment(applicationId);
  await decisionMaking(applicationId);
  await customerNotification(applicationId);
}

// case 2：内部资金管理
@DBOS.workflow()
async function liquidityManagement(branchId: string, amount: number) {
  await calculateLiquidityNeeds(branchId);
  await allocateFunds(branchId, amount);
  await updateCentralLedger(branchId, amount);
  await generateLiquidityReport(branchId, amount);
}

// case 3：核心交易处理
@DBOS.workflow()
async function realTimePayment(payment: PaymentRequest) {
  // 实时支付处理，具备完整的故障恢复能力
  await fraudDetection(payment);
  await fundsVerification(payment);
  await executePayment(payment);
  await settlementProcessing(payment);
  await complianceReporting(payment);
}

DBOS实际上是多种设计模式的巧妙组合：

1. 责任链模式 (Chain of Responsibility)

1
2
3
4
5
6
7
8
9

// 经典责任链：每个处理器处理请求或传递给下一个
@DBOS.workflow()
async function loanApproval(application: LoanApplication) {
  await documentVerification(application);  // 处理器1：文档验证
  await creditScoreCheck(application);      // 处理器2：征信检查
  await incomeVerification(application);    // 处理器3：收入验证
  await riskAssessment(application);        // 处理器4：风险评估
  await finalApproval(application);         // 处理器5：最终审批
}

2. Saga模式 (最核心的模式)

这是DBOS最准确的模式匹配：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

// Saga模式：长事务的分解与补偿
@DBOS.workflow()
async function transferMoney(transfer: MoneyTransfer) {
  try {
    // 正向操作序列
    await debitSourceAccount(transfer.from, transfer.amount);
    await creditTargetAccount(transfer.to, transfer.amount);
    await recordTransaction(transfer);
    await sendNotification(transfer);
  } catch (error) {
    // 补偿操作序列（回滚）
    await creditSourceAccount(transfer.from, transfer.amount);
    await cancelTransaction(transfer);
    throw error;
  }
}

3. 命令模式 + 备忘录模式

1
2
3
4
5
6
7

// 每个步骤都是可存储和重放的命令
@DBOS.step()
async function executePayment(payment: Payment) {
  // DBOS在PostgreSQL中存储此步骤的状态
  // 失败时可从此检查点重放
  return await processPayment(payment);
}

4. 状态机模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// 银行交易状态机
@DBOS.workflow()
async function paymentStateMachine(payment: Payment) {
  // 状态转换：待处理 -> 验证中
  await validatePayment(payment);
  
  // 状态转换：验证中 -> 资金冻结
  await freezeFunds(payment);
  
  // 状态转换：资金冻结 -> 执行中
  await executePayment(payment);
  
  // 状态转换：执行中 -> 已完成
  await completePayment(payment);
}

5. Oracle Database - transaction procedure

核心相似性分析

1. 事务边界管理

1
2
3
4
5
6
7
8

-- Oracle: 显式事务控制
BEGIN
  SAVEPOINT sp1;
  -- operations
  COMMIT;
EXCEPTION 
  WHEN OTHERS THEN ROLLBACK TO sp1;
END;

2. 状态持久化

1
2
3
4
5
6
7

-- Oracle: 数据库状态 + 过程状态
CREATE TABLE workflow_state (
  workflow_id NUMBER,
  step_name VARCHAR2(100),
  step_status VARCHAR2(20),
  step_data CLOB
);

3. 错误处理与补偿

1
2
3
4
5
6
7

-- Oracle: 异常处理
EXCEPTION
  WHEN custom_exception THEN
    -- 补偿逻辑
    ROLLBACK TO savepoint;
    -- 清理操作
END;

4. 批处理与队列处理

1
2
3
4
5
6
7
8
9
10

-- Oracle: 批处理存储过程
CREATE OR REPLACE PROCEDURE PROCESS_DAILY_BATCH AS
  CURSOR c_transactions IS 
    SELECT * FROM pending_transactions;
BEGIN
  FOR rec IN c_transactions LOOP
    PROCESS_SINGLE_TRANSACTION(rec.txn_id);
  END LOOP;
  COMMIT;
END;

模式对比表

• 分布式执行能力
• 自动故障恢复
• 云原生架构友好
• 更简洁的错误处理