Kotlin协程替代回调

折腾了好几天终于搞定了，核心其实不难，但有几个坑容易踩。把要点列出来，省得大家再走弯路。

先搞清楚原理

先说基本原理。Kotlin的核心思路是内存管理模型。简单来说就是：把输入经过一系列变换，最终得到想要的结果。这个过程可以用下面的代码来理解：

from abc import ABC, abstractmethod
from typing import List, Callable

class SingletonPattern:
    pass

class EventManager:
    def __init__(self):
        self._listeners: dict = {}

    def subscribe(self, event_type: str, listener: Callable):
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(listener)

    def unsubscribe(self, event_type: str, listener: Callable):
        if event_type in self._listeners:
            self._listeners[event_type].remove(listener)

    def notify(self, event_type: str, data: dict):
        for listener in self._listeners.get(event_type, []):
            listener(data)

manager = EventManager()
manager.subscribe('user_login', lambda d: print(f'用户 {d["name"]} 已登录'))
manager.notify('user_login', {'name': 'Alice'})

实际怎么用

实际用的时候有几个注意点。首先是配置问题，不同环境参数可能不一样。其次是性能，数据量大的时候要注意内存安全。看这段实际运行的代码：

from dataclasses import dataclass
from typing import Generic, TypeVar, Optional

T = TypeVar('T')

@dataclass
class Result(Generic[T]):
    value: Optional[T] = None
    error: Optional[str] = None
    success: bool = True

    @classmethod
    def ok(cls, value: T) -> 'Result[T]':
        return cls(value=value, success=True)

    @classmethod
    def fail(cls, error: str) -> 'Result[T]':
        return cls(error=error, success=False)

    def unwrap(self) -> T:
        if not self.success:
            raise ValueError(f'Result is error: {self.error}')
        return self.value

result = Result.ok(42)
if result.success:
    print(f'结果: {result.unwrap()}')
else:
    print(f'错误: {result.error}')

进阶一点的用法

有些场景下Kotlin的常规写法不够用，需要做内存安全。这个改动涉及的地方不多，但要注意顺序：

容易踩的坑

总结几个实际项目中的教训：Kotlin在内存安全上容易出问题。建议写单元测试覆盖这些边界情况，免得线上出故障才后悔。

写在最后

Kotlin的要点就这些。核心是内存安全，其他都是围绕这个展开的。代码已经贴在前面了，照着跑一遍基本就能上手。后续有新的发现再更新。