OCaml 教程 / Effect Handlers（OCaml 5）
Effect Handlers（OCaml 5）
Effect handlers 是 OCaml 5 引入的核心新特性，源自代数效应（algebraic effects）理论。它允许程序员定义和处理"效果"（side effects），以结构化的方式管理 I/O、状态、异常、并发等，同时保持代数性质和可组合性。
基本概念
效果（Effect）的定义
效果通过 effect 关键字定义，类似于类型构造器但用于控制流：
(* 定义一个效果 *)
type _ Effect.t +=
| Read : string Effect.t
| Write : string -> unit Effect.t
| Yield : unit Effect.t
perform：触发效果
(* 使用 perform 触发效果 *)
let read_input () : string = Effect.perform Read
let write_output s : unit = Effect.perform (Write s)
let yield () : unit = Effect.perform Yield
处理效果（Effect Handler）
使用 match_with 和效应处理函数来捕获和处理效果：
open Effect
open Effect.Deep
(* 处理 Read 效果 *)
let with_mock_input input f =
match_with f ()
{ retc = (fun x -> x); (* 正常返回 *)
exnc = (fun e -> raise e); (* 异常 *)
effc = fun (type a) (eff : a t) ->
match eff with
| Read -> Some (fun (k : (a, _) continuation) ->
continue k input (* 提供模拟输入 *)
)
| Write s -> Some (fun (k : (a, _) continuation) ->
Printf.printf "Output: %s\n" s;
continue k ()
)
| _ -> None (* 未处理的效果，向上传播 *)
}
(* 使用 *)
let program () =
let name = Effect.perform Read in
Effect.perform (Write ("Hello, " ^ name ^ "!"))
let () =
with_mock_input "World" program
(* 输出: Output: Hello, World! *)
深处理 vs 浅处理
Deep Handler（深处理）
深处理（Effect.Deep）会自动包装子计算中的效果：
open Effect.Deep
(* 深处理：自动处理嵌套调用中的效果 *)
let run_state init f =
let state = ref init in
match_with f ()
{ retc = (fun x -> (x, !state));
exnc = (fun e -> raise e);
effc = fun (type a) (eff : a t) ->
match eff with
| Get -> Some (fun (k : (a, _) continuation) ->
continue k !state
)
| Set v -> Some (fun (k : (a, _) continuation) ->
state := v;
continue k ()
)
| _ -> None
}
Shallow Handler（浅处理）
浅处理（Effect.Shallow）需要手动调用 continue 来递归处理：
open Effect.Shallow
let with_log f =
continue_with (fiber f) ()
{ retc = (fun x -> Printf.printf "Done: %d\n" x; x);
exnc = (fun e -> raise e);
effc = fun (type a) (eff : a t) ->
match eff with
| Log msg -> Some (fun (k : (a, _) continuation) ->
Printf.printf "[LOG] %s\n" msg;
continue_with k ()
{ retc = (fun x -> x);
exnc = (fun e -> raise e);
effc = (fun _ -> None)
}
)
| _ -> None
}
⚠️ 注意：浅处理的 continuation 只能使用一次。多次 continue 会导致运行时错误。深处理由运行时自动管理。
类型系统中的效果
OCaml 5 的类型系统追踪效果，确保效果被处理：
(* 效果类型标注 *)
let read_input () : string = Effect.perform Read
(* 类型: unit -> string
但实际需要在有 Read handler 的上下文中运行 *)
(* 未处理的效果会在类型层面被捕获 *)
(* 如果调用 perform 但没有 handler，编译器会警告 *)
效果在类型系统中的表现：
| 表达式 | 类型 | 效果 |
|---|---|---|
42 | int | 无 |
Effect.perform Read | string | Read |
print_endline "hi" | unit | IO |
💡 提示：OCaml 5.0-5.2 的效果追踪还不完善。未来版本会加强效果系统在类型层面的约束。
常见效果模式
State 效果
open Effect
open Effect.Deep
type _ Effect.t +=
| Get : int Effect.t
| Set : int -> unit Effect.t
let get () : int = Effect.perform Get
let set (v : int) : unit = Effect.perform (Set v)
(* 计数器程序 *)
let counter () =
let n = get () in
set (n + 1);
let m = get () in
Printf.printf "Count: %d\n" m
(* 运行带状态的程序 *)
let run_with_state init f =
let state = ref init in
match_with f ()
{ retc = (fun x -> x);
exnc = (fun e -> raise e);
effc = fun (type a) (eff : a t) ->
match eff with
| Get -> Some (fun (k : (a, _) continuation) ->
continue k !state
)
| Set v -> Some (fun (k : (a, _) continuation) ->
state := v;
continue k ()
)
| _ -> None
}
let () = run_with_state 0 counter
(* 输出: Count: 1 *)
Exception 效果
open Effect
open Effect.Deep
type _ Effect.t +=
| Raise : string -> 'a Effect.t
let raise_error msg = Effect.perform (Raise msg)
(* 使用异常效果 *)
let risky_computation () =
let x = 42 in
if x > 10 then raise_error "Value too large";
x
(* 处理异常效果 *)
let catch f handler =
match_with f ()
{ retc = (fun x -> Ok x);
exnc = (fun e -> Error (Printexc.to_string e));
effc = fun (type a) (eff : a t) ->
match eff with
| Raise msg -> Some (fun (k : (a, _) continuation) ->
handler msg
)
| _ -> None
}
let () =
match catch risky_computation (fun msg ->
Printf.printf "Caught: %s\n" msg; -1
) with
| Ok n -> Printf.printf "Result: %d\n" n
| Error msg -> Printf.printf "Error: %s\n" msg
Concurrent 效果
open Effect
open Effect.Deep
type _ Effect.t +=
| Fork : (unit -> unit) -> unit Effect.t
| Yield : unit Effect.t
let fork f = Effect.perform (Fork f)
let yield () = Effect.perform Yield
(* 简单的协作式调度器 *)
let run_scheduler main =
let queue = Queue.create () in
let enqueue k = Queue.push k queue in
let rec step () =
if not (Queue.is_empty queue) then
let k = Queue.pop queue in
continue k ()
in
match_with main ()
{ retc = (fun () -> step ());
exnc = (fun e -> raise e);
effc = fun (type a) (eff : a t) ->
match eff with
| Fork f -> Some (fun (k : (a, _) continuation) ->
enqueue k;
match_with f ()
{ retc = (fun () -> step ());
exnc = (fun e -> raise e);
effc = (fun _ -> None) }
)
| Yield -> Some (fun (k : (a, _) continuation) ->
enqueue k; step ())
| _ -> None
}
let () =
run_scheduler (fun () ->
Printf.printf "Task 1 start\n";
yield ();
Printf.printf "Task 1 end\n"
)
效果与回调（CPS）对比
| 特性 | Effect Handlers | CPS（回调） | Lwt/Async |
|---|---|---|---|
| 代码风格 | 同步（看起来像直接风格） | 回调嵌套 | bind/let* 链 |
| 栈管理 | 运行时管理 continuation | 手动栈 | 手动 promise 链 |
| 类型侵入 | 无（返回原始类型） | 无 | 需要 Lwt.t 包装 |
| 性能 | continuation 捕获有开销 | 低开销 | 低开销 |
| 可组合性 | 优秀（handler 栈） | 一般 | 一般 |
(* Effect handler 风格 — 看起来像同步代码 *)
let process () =
let input = Effect.perform Read in (* 像直接返回 *)
let result = transform input in
Effect.perform (Write result)
(* Lwt 风格 — 需要 bind *)
let process () =
let* input = read () in
let result = transform input in
write result
💡 提示：Effect handlers 的主要优势是代码可读性——异步/并发代码可以写成同步风格，由运行时管理暂停和恢复。
效果系统的形式语义
Effect handlers 基于代数效应理论，其核心性质：
- 代数律：效果满足代数等式
(* Get-Set 律 *)
set v; get () ≡ set v; return v
(* Set-Set 律 *)
set u; set v ≡ set v
- 组合性：多个 handler 可以嵌套
(* 状态 handler + 日志 handler 嵌套 *)
let run_both f =
run_with_state 0 (fun () ->
run_with_log (fun () ->
f ()
)
)
- 动态绑定：handler 作用域是动态的，取决于调用栈
ocaml-multicore 运行时
Effect handlers 的运行时基础设施：
┌────────────────────────────────────┐
│ OCaml 5 Runtime │
├────────────────────────────────────┤
│ Domain 0 │ Domain 1 │ ... │
│ ┌────────┐ │ ┌────────┐ │ │
│ │ Fiber │ │ │ Fiber │ │ │
│ │ Stack │ │ │ Stack │ │ │
│ └────────┘ │ └────────┘ │ │
│ ┌────────┐ │ ┌────────┐ │ │
│ │ Handler│ │ │ Handler│ │ │
│ └────────┘ │ └────────┘ │ │
├─────────────┴─────────────┴───────┤
│ Systhreads / GC │
└────────────────────────────────────┘
每个 fiber 有独立的栈空间，continuation 捕获当前栈帧。
效果的最佳实践
1. 定义清晰的效果类型
(* ✅ 定义在独立模块中 *)
module Eff = struct
type _ Effect.t += Read : string Effect.t
type _ Effect.t += Write : string -> unit Effect.t
type _ Effect.t += Abort : string -> 'a Effect.t
end
2. 提供组合式 handler
(* ✅ handler 可以组合 *)
let with_state init f = (* ... *)
let with_log f = (* ... *)
let with_abort f = (* ... *)
(* 组合使用 *)
let run program =
with_state 0 (fun () ->
with_log (fun () ->
with_abort (fun () ->
program ()
)
)
)
3. 避免在热路径中使用 perform
(* ❌ 每次循环都 perform *)
let bad_loop () =
for i = 0 to 1_000_000 do
Effect.perform (Set i)
done
(* ✅ 批量操作 *)
let good_loop () =
let state = ref 0 in
for i = 0 to 1_000_000 do
state := i
done;
Effect.perform (Set !state)
⚠️ 注意：Effect.perform 需要捕获 continuation（保存栈），有一定开销。不要在紧密循环中使用。
4. 错误处理
(* ✅ 使用 Effect 进行结构化错误处理 *)
type _ Effect.t += Fail : string -> 'a Effect.t
let safe_run f =
match_with f ()
{ retc = (fun x -> Ok x);
exnc = (fun e -> Error (Printexc.to_string e));
effc = fun (type a) (eff : a t) ->
match eff with
| Fail msg -> Some (fun (_ : (a, _) continuation) ->
Error msg
)
| _ -> None
}
业务场景
场景：数据库事务管理
type _ Effect.t +=
| Begin_tx : unit Effect.t
| Commit : unit Effect.t
| Rollback : unit Effect.t
| Query : string -> string list Effect.t
let with_transaction f =
Effect.perform Begin_tx;
match Effect.perform (Query "SAVEPOINT sp") with
| _ ->
(try
let result = f () in
Effect.perform Commit;
result
with e ->
Effect.perform Rollback;
raise e)
场景：可测试的 IO
(* 业务逻辑使用效果，不直接依赖 IO *)
let business_logic () =
let data = Effect.perform Read in
let result = process data in
Effect.perform (Write result)
(* 测试时用 mock handler *)
let test () =
let output = ref "" in
match_with business_logic ()
{ retc = (fun x -> x);
exnc = (fun e -> raise e);
effc = fun (type a) (eff : a t) ->
match eff with
| Read -> Some (fun k -> continue k "test data")
| Write s -> Some (fun k -> output := s; continue k ())
| _ -> None
};
assert (!output = "expected")