『状态』驱动的世界：ReactiveCocoa

原文

请您阅读原文 『状态』驱动的世界：ReactiveCocoa,作者 Draven。声明：本文只用做RAC相关知识点梳理，不做他用。

RAC 设计思维

去中心化的方式，能简化整个系统的构造，使得各个组件只需要关心状态，以及状态对应的动作；不再需要一个中枢系统来组织、管理其它的组件，并负责大多数的业务逻辑。这种自底向下的、状态驱动的构建方式能够使用多个较小的组件，减少臃肿的中枢出现的可能性，从而降低系统的复杂度。

ReactiveCocoa 对于状态的理解与上述，将原有的各种设计模式，包括代理、Target/Action、通知中心以及观察者模式各种『输入』，都抽象成了信号（也可以理解为状态流）让单一的组件能够对自己的响应动作进行控制，简化了视图控制器的负担。

RACSignal 简介

RACSignal 其实是抽象类 RACStream 的子类，在整个 ReactiveObjc 工程中有另一个类 RACSequence 也继承自抽象类 RACStream

RACSignal-Hierachy.png

RACSignal 可以说是 ReactiveCocoa 中的核心类，也是最重要的概念，整个框架围绕着 RACSignal 的概念进行组织，对 RACSignal 最简单的理解就是它表示一连串的状态：

What-is-RACSignal.png

在状态改变时，对应的订阅者 RACSubscriber 就会收到通知执行相应的指令，在 ReactiveCocoa 的世界中所有的消息都是通过信号的方式来传递的，原有的设计模式都会简化为一种模型。

RACStream

RACStream 作为抽象类本身不提供方法的实现，其实现内部原生提供的而方法都是抽象方法，会在调用时直接抛出异常：

+ (__kindof RACStream *)empty {
    NSString *reason = [NSString stringWithFormat:@"%@ must be overridden by subclasses", NSStringFromSelector(_cmd)];
    @throw [NSException exceptionWithName:NSInternalInconsistencyException reason:reason userInfo:nil];
}

- (__kindof RACStream *)bind:(RACStreamBindBlock (^)(void))block;
+ (__kindof RACStream *)return:(id)value;
- (__kindof RACStream *)concat:(RACStream *)stream;
- (__kindof RACStream *)zipWith:(RACStream *)stream;

RACStream-AbstractMethod

上面的这些抽象方法都需要子类覆写，不过 RACStream 在 Operations 分类中使用上面的抽象方法提供了丰富的内容，比如说 -flattenMap: 方法：

- (__kindof RACStream *)flattenMap:(__kindof RACStream * (^)(id value))block {
    Class class = self.class;

    return [[self bind:^{
        return ^(id value, BOOL *stop) {
            id stream = block(value) ?: [class empty];
            NSCAssert([stream isKindOfClass:RACStream.class], @"Value returned from -flattenMap: is not a stream: %@", stream);

            return stream;
        };
    }] setNameWithFormat:@"[%@] -flattenMap:", self.name];
}

其他方法比如-skip:、-take:、-ignore：等等实例方法都构建在这些抽象方法之上，只要子类覆写了所有抽象方法就能自动获得所有的 Operation 分类中的方法。

RACStream-Operation

信号的创建过程十分简单，-createSignal: 是推荐的创建信号的方法，方法其实只做了一次转发：

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe];
}

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    RACDynamicSignal *signal = [[self alloc] init];
    signal->_didSubscribe = [didSubscribe copy];
    return [signal setNameWithFormat:@"+createSignal:"];
}

该方法其实只是创建了一个 RACDynamicSignal 实例并保存了传入的 didSubscribe 代码块，在每次有订阅者订阅当前信号时，都会执行一遍，向订阅者发送消息。

RACSignal 类簇

虽然 -createSignal: 的方法签名上返回的是 RACSignal 对象的实例，但是实际上这里返回的是 RACDynamicSignal，也就是 RACSignal 的子类；同样，在 ReactiveCocoa 中也有很多其他的 RACSignal 子类。

使用类簇的方式设计的 RACSignal 在创建实例时可能会返回 RACDynamicSignal、RACEmptySignal、RACErrorSignal 和 RACReturnSignal 对象：

RACSignal-Subclasses

其实这几种子类并没有对原有的 RACSignal 做出太大的改变，它们的创建过程也不是特别的复杂，只需要调用 RACSignal 不同的类方法：

RACSignal-Instantiate-Object

RACSignal 只是起到了一个代理的作用，最后的实现过程还是会指向对应的子类：

+ (RACSignal *)error:(NSError *)error {
    return [RACErrorSignal error:error];
}

+ (RACSignal *)empty {
    return [RACEmptySignal empty];
}

+ (RACSignal *)return:(id)value {
    return [RACReturnSignal return:value];
}

RAC 信号订阅流程

RACSignal-Subcribe-Process

创建信号 RACSignal ，RACSignal 调用 -subscribeNext: 方法返回一个 RACDisposable，在订阅过程中生成了一个 RACSubscriber 对象，向这个对象发送消息 -sendNext: 时，就会向所有的订阅者发送消息。

信号的订阅

RACSignal-Subscribe-Methods.png

订阅者可以选择自己想要感兴趣的信息类型 next/error/completed 进行关注，并在对应的信息发生时调用 block 进行处理回调。

所有的方法其实只是对 nextBlock、completedBlock 以及 errorBlock 的组合，这里以其中最长的 -subscribeNext:error:completed: 方法的实现为例（也只需要介绍这一个方法）：

- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
    return [self subscribe:o];
}

方法中传入的所有 block 参数都应该是非空的。
拿到了传入的 block 之后，使用 +subscriberWithNext:error:completed: 初始化一个 RACSubscriber 对象的实例：

+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
    RACSubscriber *subscriber = [[self alloc] init];

    subscriber->_next = [next copy];
    subscriber->_error = [error copy];
    subscriber->_completed = [completed copy];

    return subscriber;
}

在拿到这个对象之后，调用 RACSignal 的 -subscribe: 方法传入订阅者对象：

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCAssert(NO, @"This method must be overridden by subclasses");
    return nil;
}

RACSignal 类中其实并没有实现这个实例方法，需要在上文提到的四个子类对这个方法进行覆写，这里仅分析 RACDynamicSignal 中的方法：

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

    RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
        RACDisposable *innerDisposable = self.didSubscribe(subscriber);
        [disposable addDisposable:innerDisposable];
    }];

    [disposable addDisposable:schedulingDisposable];
    
    return disposable;
}

RACPassthroughSubscriber 就像它的名字一样，只是对上面创建的订阅者对象进行简单的包装，将所有的消息转发给内部的 innerSubscriber，也就是传入的 RACSubscriber 对象：

- (instancetype)initWithSubscriber:(id<RACSubscriber>)subscriber signal:(RACSignal *)signal disposable:(RACCompoundDisposable *)disposable {
    self = [super init];

    _innerSubscriber = subscriber;
    _signal = signal;
    _disposable = disposable;

    [self.innerSubscriber didSubscribeWithDisposable:self.disposable];
    return self;
}

如果直接简化 -subscribe:方法的实现，你可以看到一个看起来极为敷衍的代码：

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    return self.didSubscribe(subscriber);
}

总而言之，信号的订阅过程就是初始化 RACSubscriber 对象，然后执行 didSubscribe 代码块的过程生成 disposable。

Principle-of-Subscribing-Signals

信息的发送

在 RACSignalBindBlock 中，订阅者可以根据自己的兴趣选择自己想要订阅哪种消息；我们也可以按需发送三种消息：

RACSignal-Subscription-Messages-Sending

订阅的回收过程

在创建信号时，我们向 -createSignal:方法中传入了 didSubscribe 信号，这个 block 在执行结束时会返回一个 RACDisposable 对象，用于在订阅结束时进行必要的清理，同样也可以用于取消因为订阅创建的正在执行的任务。

而处理这些事情的核心类就是 RACDisposable 以及它的子类：

RACDisposable-And-Subclasses

这篇文章中主要关注的是左侧的三个子类，当然 RACDisposable 的子类不止这三个，还有用于处理 KVO 的 RACKVOTrampoline，不过在这里我们不会讨论这个类的实现。

RACDisposable

在继续分析讨论订阅的回收过程之前，笔者想先对 RACDisposable 进行简要的剖析和介绍：

RACDisposable.png

类 RACDisposable 是以 _disposeBlock 为核心进行组织的，几乎所有的方法以及属性其实都是对 _disposeBlock 进行的操作。

关于 _disposeBlock 中的 self

这一小节的内容是可选的，跳过不影响整篇文章阅读的连贯性。

_disposeBlock 是一个私有的指针变量，当 void (^)(void) 类型的 block 被传入之后都会转换成 CoreFoundation 中的类型并以 void * 的形式存入 _disposeBlock 中：

+ (instancetype)disposableWithBlock:(void (^)(void))block {
    return [[self alloc] initWithBlock:block];
}

- (instancetype)initWithBlock:(void (^)(void))block {
    self = [super init];

    _disposeBlock = (void *)CFBridgingRetain([block copy]); 
    OSMemoryBarrier();

    return self;
}

奇怪的是，_disposeBlock 中不止会存储代码块 block，还有可能存储桥接之后的self ：

- (instancetype)init {
    self = [super init];

    _disposeBlock = (__bridge void *)self;
    OSMemoryBarrier();

    return self;
}

这里，刚开始看到可能会觉得比较奇怪，有两个疑问需要解决：

• 为什么要提供一个 -init 方法来初始化 RACDisposable 对象？
• 为什么要向_disposeBlock 中传入当前对象？

对于 RACDisposable 来说，虽然一个不包含 _disposeBlock 的对象没什么太多的意义，但是对于 RACSerialDisposable 等子类来说，却不完全是这样，因为 RACSerialDisposable 在 -dispose 时，并不需要执行 disposeBlock，这样就浪费了内存和 CPU 时间；但是同时我们需要一个合理的方法准确地判断当前对象的 isDisposed：

所以，使用向 _disposeBlock 中传入 NULL 的方式来判断 isDisposed；在 -init 调用时传入 self 而不是 NULL 防止状态被误判，这样就在不引入其他实例变量、增加对象的设计复杂度的同时，解决了这两个问题。

如果仍然不理解上述的两个问题，在这里举一个错误的例子，如果 _disposeBlock 在使用时只传入 NULL 或者 block，那么在 RACCompoundDisposable 初始化时，是应该向 _disposeBlock 中传入什么呢？

• 传入 NULL 会导致在初始化之后 isDisposed == YES，然而当前对象根本没有被回收；
• 传入 block 会导致无用的 block 的执行，浪费内存以及 CPU 时间；

这也就是为什么要引入 self 来作为 _disposeBlock 内容的原因。

-dispose: 方法的实现

这个只有不到 20 行的 -dispose: 方法已经是整个 RACDisposable 类中最复杂的方法了：

- (void)dispose {
    void (^disposeBlock)(void) = NULL;

    while (YES) {
        void *blockPtr = _disposeBlock;
        if (OSAtomicCompareAndSwapPtrBarrier(blockPtr, NULL, &_disposeBlock)) {
            if (blockPtr != (__bridge void *)self) {
                disposeBlock = CFBridgingRelease(blockPtr);
            }

            break;
        }
    }

    if (disposeBlock != nil) disposeBlock();
}

RACSerialDisposable

RACSerialDisposable 是一个用于持有 RACDisposable 的容器，它一次只能持有一个RACDisposable 的实例，并可以原子地换出容器中保存的对象：

- (RACDisposable *)swapInDisposable:(RACDisposable *)newDisposable {
    RACDisposable *existingDisposable;
    BOOL alreadyDisposed;

    pthread_mutex_lock(&_mutex);
    alreadyDisposed = _disposed;
    if (!alreadyDisposed) {
        existingDisposable = _disposable;
        _disposable = newDisposable;
    }
    pthread_mutex_unlock(&_mutex);

    if (alreadyDisposed) {
        [newDisposable dispose];
        return nil;
    }

    return existingDisposable;
}

线程安全的 RACSerialDisposable 使用 pthred_mutex_t 互斥锁来保证在访问关键变量时不会出现线程竞争问题。
-dispose 方法的处理也十分简单：

- (void)dispose {
    RACDisposable *existingDisposable;

    pthread_mutex_lock(&_mutex);
    if (!_disposed) {
        existingDisposable = _disposable;
        _disposed = YES;
        _disposable = nil;
    }
    pthread_mutex_unlock(&_mutex);
    
    [existingDisposable dispose];
}

使用锁保证线程安全，并在内部的 _disposable 换出之后在执行 -dispose 方法对订阅进行处理。

RACCompoundDisposable

与 RACSerialDisposable 只负责一个 RACDisposable 对象的释放不同；RACCompoundDisposable 同时负责多个 RACDisposable 对象的释放。

相比于只管理一个 RACDisposable 对象的 RACSerialDisposable，RACCompoundDisposable 由于管理多个对象，其实现更加复杂，而且为了性能和内存占用之间的权衡，其实现方式是通过持有两个实例变量：

@interface RACCompoundDisposable () {
    ...
    RACDisposable *_inlineDisposables[RACCompoundDisposableInlineCount];

    CFMutableArrayRef _disposables;
    ...
}

在对象持有的 RACDisposable 不超过 RACCompoundDisposableInlineCount 时，都会存储在_inlineDisposables 数组中，而更多的实例都会存储在 _disposables 中：

RACCompoundDisposable

RACCompoundDisposable 在使用 -initWithDisposables:初始化时，会初始化两个 RACDisposable 的位置用于加速销毁订阅的过程，同时为了不浪费内存空间，在默认情况下只占用两个位置：

- (instancetype)initWithDisposables:(NSArray *)otherDisposables {
    self = [self init];

    [otherDisposables enumerateObjectsUsingBlock:^(RACDisposable *disposable, NSUInteger index, BOOL *stop) {
        self->_inlineDisposables[index] = disposable;
        if (index == RACCompoundDisposableInlineCount - 1) *stop = YES;
    }];

    if (otherDisposables.count > RACCompoundDisposableInlineCount) {
        _disposables = RACCreateDisposablesArray();

        CFRange range = CFRangeMake(RACCompoundDisposableInlineCount, (CFIndex)otherDisposables.count - RACCompoundDisposableInlineCount);
        CFArrayAppendArray(_disposables, (__bridge CFArrayRef)otherDisposables, range);
    }

    return self;
}

如果传入的 otherDisposables 多于 RACCompoundDisposableInlineCount，就会创建一个新的 CFMutableArrayRef 引用，并将剩余的 RACDisposable 全部传入这个数组中。

在 RACCompoundDisposable 中另一个值得注意的方法就是 -addDisposable:

- (void)addDisposable:(RACDisposable *)disposable {
    if (disposable == nil || disposable.disposed) return;

    BOOL shouldDispose = NO;

    pthread_mutex_lock(&_mutex);
    {
        if (_disposed) {
            shouldDispose = YES;
        } else {
            for (unsigned i = 0; i < RACCompoundDisposableInlineCount; i++) {
                if (_inlineDisposables[i] == nil) {
                    _inlineDisposables[i] = disposable;
                    goto foundSlot;
                }
            }

            if (_disposables == NULL) _disposables = RACCreateDisposablesArray();
            CFArrayAppendValue(_disposables, (__bridge void *)disposable);
        foundSlot:;
        }
    }
    pthread_mutex_unlock(&_mutex);
    if (shouldDispose) [disposable dispose];
}

在向 RACCompoundDisposable 中添加新的 RACDisposable 对象时，会先尝试在 _inlineDisposables 数组中寻找空闲的位置，如果没有找到，就会加入到 _disposables 中；但是，在添加 RACDisposable 的过程中也难免遇到当前 RACCompoundDisposable 已经 dispose 的情况，而这时就会直接-dispose` 刚刚加入的对象。