对于订阅和反订阅的一些思考

前言

因为项目的一些历史原因，最近在进行重构，碰到了一些比较核心的订阅部分，写出来记录一下。

订阅这个概念我是首先在RAC里见到。在响应式编程里一切都是数据流，RAC在实现上创建了信号和订阅的概念，你可以监听信号的变化，iOS底层是通过KVO和运行时实现的。KVO大家应该比较熟悉，网上说它原理的文章太多了，大致就是通过运行时动态创建子类，hook属性的set方法，在值改变的时候回调，不过我们需要自己手动移除监听，否则会存在内存问题，这个设计饱受诟病。从Swift 4.0开始，KVO有了一种新的写法。API是这样的：

1
2
3

func observe<Value>(_ keyPath: KeyPath<Observer, Value>,
  options: NSKeyValueObservingOptions = default,
  changeHandler: @escaping (Observer, NSKeyValueObservedChange<Value>) -> Void) -> NSKeyValueObservation

然后我们就可以这样写：

class TestObj: NSObject {
    @objc dynamic var str = ""

    var observer: NSKeyValueObservation?

    func testKVO() {
        self.observer = self.observe(\.str, options: .new) { (obj, values) in
            print(values)
        }
    }
}

这里需要注意的是必须创建一个强引用持有返回的对象，否则监听事件就无效了。这个对象的类型是NSKeyValueObservation，结构很简单，定义是这样的。

public class NSKeyValueObservation : NSObject {

    ///invalidate() will be called automatically when an NSKeyValueObservation is deinited
    public func invalidate()
}

注释说这个对象在释放的时候会自动解除引用，也就是说不需要我们自己手动移除KVO监听了，监听事件和对象的生命周期绑定。RAC的里的RACKVOTrampoline就是这种实现的一个例子。

目的

先说说业务场景。App在某些页面需要给服务器发送指令，请求返回相应的数据。项目里因为对数据实时性要求比较高，而且为了统一前端和移动端，使用了WebSocket协议。指令存在很多种类型，有针对单个Tab的，或者详情页中针对单个币种的，他们之间互不影响。但是和订阅一样，有订阅就有反订阅，所以在不需要的时候需要向服务器发送取消的指令。所以我希望这里的指令能像上面说的订阅模型一样，在对象销毁的时候自动发送取消指令。RAC有一套比较完整的订阅模式，但是依赖比较严重，项目中希望慢慢移除掉它，所以实现了一个更贴近业务轻量的框架。

实现

其实指令大致就分为两类，一种是tab指令，一种是单个币种的指令。单就这两个就没什么好说的了，无非就是页面出现的时候创建指令对象，发指令给服务器，页面销毁也就发送取消的指令给服务器，中断数据传输。
唯一一个复杂的点在单个币种指令的集合。因为有一种业务场景是这样的，本地允许用户添加自选，不上传服务器。那么服务器没办法知道用户的自选列表里有什么，传什么数据就需要App自己指定。当时为了赶需求，这种方式就直接使用了单个指令，有多少个自选币种，就创建多少个指令对象。虽然简单粗暴，但是在取消的时候就有一些麻烦了。后来重构，发现确实也只有这种方式可行，这和业务场景有关。比如我从本地自选A页面跳转到某个需要发送单个指令的页面B，这时候订阅对象应该是复用的，因为从B页面回到A的时候，单个指令也并不应该销毁，因为A页面依然持有它。当时为了这个问题头疼了很久，后来恍然大悟，这不就是和MRC原理差不多嘛。

明白了之后就很简单了。

首先在自选列表A页面需要发送指令的时候由一个全局的管理器创建一个指令对象给A页面持有，因为列表允许多个单指令集合，所以管理器需要把这些单个对象组合成一个集合C返回。
跳转到B页面时，同样需要通过管理器创建指令，管理器发现这个单指令存在，那么就不要创建了，返回它的引用，这时候集合C的引用计数就是2。
B页面在销毁的时候引用计数-1，但是因为A页面的持有，集合C实际上引用计数还是1，不会释放，也就不会发送取消指令。

看似可行的方案，实践中碰上了一些问题。按照MRC的规则，谁创建，谁释放，那么A页面在销毁的时候按理说集合C也要销毁，但是管理器也需要持有集合C，这样其它页面来请求指令的时候才能知道应该创建新的指令对象还是应该返回引用。但是管理器不应该强持有集合C，不然集合C就永远无法释放了。所以管理器在持有集合C的时候需要弱引用。这里用到了NSHashTable，它的用法和NSArray很相似，但是NSHashTable可以持有弱引用的对象。

所以博客到这里就写完了吗？当然没有。还有一个最大的发送取消集合C的指令的问题。服务器提供了一个批量取消单指令的指令，很适合集合C的场景。想法很简单，在集合C销毁的时候把它持有的单指令批量取消，就不需要一个个的去发送取消的指令了。那么问题来了，集合C怎么知道哪些单个指令需要取消，哪些不需要呢。而且集合C必须强持有它的单个指令，按照释放顺序，在集合C走到deinit方法的时候，它的属性是还能访问的。所以我们需要让集合C的单指令提前释放，并且还需要知道哪些单个指令需要发送取消指令，哪些不需要。核心代码如下：

class ListSubscribeObservation {
  //单订阅的数组
  var children: [SingleSubscribeObservation]?

  deinit() {
        //持有一份单指令的弱引用 然后让单指令集合提前释放
        //如果弱引用还存在 那么说明还有地方在强应用这个单指令 那么不应该把它取消

        let obsTable = NSHashTable<SingleSubscribeObservation>(options: [.weakMemory, .objectPointerPersonality], capacity: 0)
        //要发送给服务器的指令集
        var subKeys = [String]()
        autoreleasepool {
            while let obj = self.children?.popLast() {
                obsTable.add(obj)
                //这个标记用于表示是否应该发送取消指令
                item.autoUnsub = false
                subKeys.append(obj.value)
            }
        }

        //走到这里单指令集合应该已经被提前释放了 剩下的则不应该发送指令 将它移除出数组
        for item in obsTable.objectEnumerator() {
            if let item = item as? SingleSubscribeObservation {
                //从批量释放中移除
                if subKeys.contains(item.value) {
                    //恢复单订阅自动释放
                    item.autoUnsub = true
                    subKeys.remove(item.value)
                }
            }
        }

        if subKeys.count > 0 {
          //发送订阅指令
            let ws = WebSocket.shared
            ws.removeKeys(subKeys)
        }
    }
  }

但是实际上在某些时候即使这样写，children仍然会在这个对象之后释放，所以最后的办法是在业务层加了延迟发送指令的代码。因为children最后一定会释放，它内部的单个指令也会快速释放，但是走的是单个释放的。所以做法是释放时通过定时器delay个0.1s，如果0.1s内再次走了这个方法，则把上一次的timer取消掉，否则执行批量释放，代码如下：

class WebSocket {

// 临时需要移除订阅的数组
private var tempRemoveKeys = Set<String>()

func removeKeys(_ keys: [String]) {
  if keys.count == 0 {
      return
  }

  //使用临时集合保存释放的key
  self.tempRemoveKeys.formUnion(keys)

  //短时间的批量释放合并到一起处理
  self.removeDelayTimer?.invalidate()
  self.removeDelayTimer = Timer.scheduledTimer(withTimeInterval: 0.2, block: { [weak self] (_) in
      guard let self = self else {
          return
      }

      let tempKeys = Array(self.tempRemoveKeys)
      //发送取消指令
      self.unsubKeys(tempKeys)
      //移除临时集合
      self.tempRemoveMarkets.removeAll()
      }, repeats: false)
    }
}

总结

其实这个解决方案并不算好，导致调用时机没有按照预期执行的原因有待研究，这里对思考过程做一个记录。