创建信号

RACSignal *signal = [RACSignal createSignal:^RACDisposable * _Nullable(id<RACSubscriber>  _Nonnull subscriber) {
    [subscriber sendNext:@1];
    [subscriber sendNext:@2];
     [subscriber sendCompleted];
     return [RACDisposable disposableWithBlock:^{
            NSLog(@"signal");
     }];
 }];

上面是创建信号的代码，下面看看其内部是如何实现的：

//RACSignal的createSignal方法内部实现是返回一个子类RACDynamicSignal实例对象
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe];
}

// 子类RACDynamicSignal的createSignal方法内部实例化一个RACDynamicSignal对象
//且这个对象有一个成员变量didSubscribe接收外面传来的didSubscribe，留着后面订阅信号时调用
+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    RACDynamicSignal *signal = [[self alloc] init];
    signal->_didSubscribe = [didSubscribe copy];
    return [signal setNameWithFormat:@"+createSignal:"];
}

订阅信号

 [signal subscribeNext:^(id  _Nullable x) {
        NSLog(@"%@",x);
 }];

接下来看看信号订阅内部实现：

//signal调用subscribeNext订阅信号，该方法内部创建了一个RACSubscriber实例对象
//并用该对象作为参赛调用subscribe方法
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    return [self subscribe:o];
}

//方法内部实例化一个RACSubscriber对象，该对象的三个成员变量next、error、completed分别接收三个参数
+ (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;
}

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCParameterAssert(subscriber != nil);

    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

    if (self.didSubscribe != NULL) {
        RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            //此处的self.didSubscribe就是信号创建时持有的那个block didSubscribe
            //调用didSubscribe，信号创建时的sendNext：消息被触发
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];

        [disposable addDisposable:schedulingDisposable];
    }
    
    return disposable;
}

订阅后消息被发送，那消息的接收又是如何实现的呢？下面看看sendNext：内部是如何实现的：

- (void)sendNext:(id)value {
    @synchronized (self) {
        //这个nextblock就是创建subscriber时传入的block，
       //sendNext内部就是调用这个block，这样消息接收被触发了
        void (^nextBlock)(id) = [self.next copy];
        if (nextBlock == nil) return;

        nextBlock(value);
    }
}

总结：信号RACSignal的创建实际是创建了一个子类RACDynamicSignal对象，该对象持有信号创建传入的block（didSubscribe，用来发送消息）；信号的订阅内部会创建一个RACSubscriber对象(该对象持有nextBlock,用来接收消息),然后这个subscriber对象作为参数调用subscribe方法；subscribe方法内部以subscriber作为参赛调用信号创建时的didSubscribe。这样didSubscribe内部通过sendNext方法发送消息，而sendNext内部会调用nextBlock接收消息。至此，信号的创建、订阅，消息的发送和接收全部完成。

RACDisposable类封装了删除和清理订阅所需的工作，下面是其内部实现：

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

- (instancetype)initWithBlock:(void (^)(void))block {
    NSCParameterAssert(block != nil);

    self = [super init];

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

    return self;
}

创建RACDisposable时会传入一个block，我们可以用它做一些订阅结束后的处理；当订阅者销毁或者取消订阅时这个block会触发调用；

// 取消订阅
- (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();
}

由dispose内部实现可以看出取消订阅时，只要不为空一定会调用创建时传入的block；