RACSignal的子類們

研究和使用RAC有一段時間了，但一直很懶，不想寫任何東西。決定還是寫一點東西吧，也算是強迫自己整理一下。計劃從最常用的也是RAC的核心(信號) RACSignal 這個類說起，來引出RAC的方方面面：冷熱信號實現(xiàn)及轉(zhuǎn)換、RAC中FRP的體現(xiàn)、rac中的類和協(xié)議(RACDisposable、RACSubscriber、RACCommand、 RACChannel...)、 對信號的處理和控制(combineLatest、bind...)、ReactiveCocoa的設(shè)計思想及結(jié)構(gòu)體系、遇到的一些坑...

RACSignal是什么？

在RAC中RACSignal是一個工廠類，他通過各個工廠方法生成不同的子類實例。同時他繼承基類RACStream(RAC的抽象類，只是定義了處理信號流的方法，子類通過繼承來具體實現(xiàn))。

下面具體討論RACSignal的幾個子類

1. RACDynamicSignal

用法：

//1. 創(chuàng)建信號
 RACSignal *dynamicSignal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        [subscriber sendNext:@"send next"];
        [subscriber sendCompleted];
        return [RACDisposable disposableWithBlock:^{
            NSLog(@" to release some object because signal had completed or error");
        }];
    }];
//2. 訂閱信號
 [dynamicSignal subscribeNext:^(id x) {
        NSLog(@"%@",x);
    }];

上面的代碼創(chuàng)建了一個signal， 如果需要dynamicSignal信號工作，我們需要對其訂閱；訂閱之后subscriber就會將"send next"傳遞到x中,因此NSLog的打印就是"send next"，如果你使用過RAC一定對此不會陌生。 平時我們所指的signal其實就是RACDynamicSignal（通過createSignal工廠方法創(chuàng)建的一個子類）。還有一些特殊用途的signal，例如RACErrorSignal、RACEmptySignal、RACReturnSignal都是通過工廠方法獲得其實例。

下面通過源碼逐步解析：
1 . 創(chuàng)建信號

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    // 1、創(chuàng)建RACDynamicSignal實例：signal
 RACDynamicSignal *signal = [[self alloc] init];
    // 2、signal 中保存 傳入的didSubscribe 以便后續(xù)對其調(diào)用
 signal->_didSubscribe = [didSubscribe copy];
    //tip: rac在創(chuàng)建不同實例的時候都會調(diào)用這個方法，給name屬性賦值（RACStream中唯一具體實現(xiàn)的方法），給不同對象不同的標(biāo)記用于輸出日志等
 return [signal setNameWithFormat:@"+createSignal:"];
}

2. 訂閱信號

// subscribeNext方法是對訂閱過程的一個封裝，便于上層調(diào)用
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    //1、創(chuàng)建訂閱者，將nextBlock參數(shù)再次傳遞
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
       //2、傳入訂閱者，進行訂閱
    return [self subscribe:o];
}

2.1創(chuàng)建訂閱者

 // 這里是對訂閱者的初步創(chuàng)建，后面還有對它的二次處理
 + (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
   //1、創(chuàng)建RACSubscriber實例：subscriber
 RACSubscriber *subscriber = [[self alloc] init];
     //2、分別記錄傳入的next、error、completed
 subscriber->_next = [next copy];
 subscriber->_error = [error copy];
 subscriber->_completed = [completed copy];

 return subscriber;
}

2.2訂閱

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

    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
  /*
   * subscriber 被賦給了一個新的RACPassthroughSubscriber實例
   * 并將原來的subscriber以及signal本身和disposable作為新的subscriber的實例對象保存
   */
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];

    if (self.didSubscribe != NULL) {
          //下列方法會執(zhí)行傳入的bock參數(shù)（因涉及到RACDisposable和RACScheduler兩大塊，不便具體解釋過程，可以自己看源碼）    
            RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
           // 這里對《1.創(chuàng)建信號》中保存的didSubscribe進行調(diào)用，并將subscriber作為block參數(shù)傳入
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];

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

思路整理：首先createSignal方法創(chuàng)建了一個信號dynamicSignal，并把傳入的block參數(shù)記錄在了dynamicSignal的didSubscribe中；然后subscribeNext方法中創(chuàng)建了一個訂閱者 o 用其屬性next保存了從subscribeNext方法傳入的block，接下來subscribeNext中調(diào)用subscribe方法并將 o 作為參數(shù)傳入；而在subscribe方法中執(zhí)行了 self.didSubscribe(subscriber) 其中subscriber就是傳入subscribe方法的o。也就是執(zhí)行了一開始保存在dynamicSignal中的didSubscribe。
也就是執(zhí)行了createSignal方法中傳入的block

 ^RACDisposable *(id<RACSubscriber> subscriber) {
        [subscriber sendNext:@"send next"];
        [subscriber sendCompleted];
        return [RACDisposable disposableWithBlock:^{
            NSLog(@" to release some object because signal had completed or error");
        }];
}

終于到了最后一步，我們只需要弄清[subscriber sendNext:@"send next"];這一方法的具體實現(xiàn)，整個RACDynamicSignal的工作原理就全部揭曉。（sendCompleted 以及 RACDisposable 相關(guān)計劃下次介紹）
sendNext的實現(xiàn)如下：

- (void)sendNext:(id)value {
    @synchronized (self) {
        void (^nextBlock)(id) = [self.next copy];
        if (nextBlock == nil) return;

        nextBlock(value);
    }
}

可以看到這里其實就是對上文保存在 o 中的 next 進行執(zhí)行，同時將傳入的 value 再次當(dāng)作 next 的參數(shù)傳入。也就是執(zhí)行了訂閱信號方法subscribeNext 時傳入的 block ，如下

^(id x) {
        NSLog(@"%@",x);
    }

直到這里，整個信號的 “創(chuàng)建-發(fā)送-訂閱” 的過程全部分析完畢。
這里強調(diào)一下：RACDynamicSignal 中的信號需要被訂閱，然后信號才會被激活。

2. RACSubject

RACSubject 是 RACSignal 的一個子類，用法如下：

    //1、創(chuàng)建實例
    RACSubject *subject = [RACSubject subject];
    //2、訂閱信號
    [subject subscribeNext:^(id x) {
        NSLog(@"%@",x);
    }];
    //3、發(fā)送信號
    [subject sendNext:@"send next"];

和RACSignal不同的是 RACSubject 是先訂閱信號，再發(fā)送信號。RACSignal是冷信號，而RACSubject及其子類是熱信號，這也是為什么RACSubject的創(chuàng)建不是通過RACSignal的工廠方法而是單獨拿出來做為一個體系的原因。下面通過源碼具體分析：
1 . 創(chuàng)建信號

//僅僅只是實例了一個對象
+ (instancetype)subject {
    return [[self alloc] init];
}

2. 訂閱信號

//執(zhí)行的還是其父類RACSignal中的方法
- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    NSCParameterAssert(nextBlock != NULL);
    
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    return [self subscribe:o];
}

2.1 創(chuàng)建訂閱者

// 這里和RACSignal中也是一樣，執(zhí)行的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;
}

2.2 訂閱

// 這里開始不同，RACSubject對其進行了重寫。其實這個方法，在RACSignal中沒有實現(xiàn)，RACDynamicSignal也是執(zhí)行的RACDynamicSignal中重寫的方法
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCParameterAssert(subscriber != nil);

    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
      // 這里和RACDynamicSignal中不同，RACSubject持有一個數(shù)組subscribers，來儲存所有的訂閱者，對subject的每一次訂閱都會將訂閱者加入該數(shù)組保存起來
    NSMutableArray *subscribers = self.subscribers;
    @synchronized (subscribers) {
        [subscribers addObject:subscriber];
    }
    
    return [RACDisposable disposableWithBlock:^{
        @synchronized (subscribers) {
            // Since newer subscribers are generally shorter-lived, search
            // starting from the end of the list.
            NSUInteger index = [subscribers indexOfObjectWithOptions:NSEnumerationReverse passingTest:^ BOOL (id<RACSubscriber> obj, NSUInteger index, BOOL *stop) {
                return obj == subscriber;
            }];

            if (index != NSNotFound) [subscribers removeObjectAtIndex:index];
        }
    }];
}

3. 發(fā)送信號

// 循環(huán)遍歷subscribers數(shù)組，逐個執(zhí)行:[subscriber sendNext:value]
- (void)sendNext:(id)value {
    [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
        [subscriber sendNext:value];
    }];
}
- (void)enumerateSubscribersUsingBlock:(void (^)(id<RACSubscriber> subscriber))block {
    NSArray *subscribers;
    @synchronized (self.subscribers) {
        subscribers = [self.subscribers copy];
    }

    for (id<RACSubscriber> subscriber in subscribers) {
        block(subscriber);
    }
}

和RACSignal不同的是：RACSubject 通過subscribeNext方法將訂閱者保存在subscribers數(shù)組中，供sendNext方法調(diào)用；而RACSignal是先保存一個didSubscribe對象，然后通過subscribeNext方法創(chuàng)建一個訂閱者subscriber作為參數(shù)傳入didSubscribe中，進而subscriber才能發(fā)送信號。
簡言之：RACSubject會保存所有subscriber（狀態(tài)），sendNext只是對其操作。RACSignal 在被訂閱后才會生成subscriber，并立即對其操作。
這也是冷信號和熱信號的區(qū)別（冷信號需要通過訂閱來激活）

3. RACReplaySubject

RACReplaySubject 是 RACSubject 的一個子類，他們都是熱信號，那么他們有什么不同呢，還是從源碼入手

 //1、創(chuàng)建實例
    RACReplaySubject *subject = [RACReplaySubject replaySubjectWithCapacity:2];
    //2、訂閱信號
    [subject subscribeNext:^(id x) {
        NSLog(@"%@",x);
    }];
    //3、發(fā)送信號
    [subject sendNext:@"send next"];

1 . 創(chuàng)建信號

+ (instancetype)replaySubjectWithCapacity:(NSUInteger)capacity {
    return [(RACReplaySubject *)[self alloc] initWithCapacity:capacity];
}
- (instancetype)initWithCapacity:(NSUInteger)capacity {
    self = [super init];
    if (self == nil) return nil;
    // 這里會用一個屬性capacity 將 傳入的數(shù)記錄下來，下面會有這個數(shù)的用途
    _capacity = capacity;
    _valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
    
    return self;
}
// 也可以直接對其實例，capacity為最大無符號整數(shù)
- (instancetype)init {
        // const NSUInteger RACReplaySubjectUnlimitedCapacity = NSUIntegerMax;
    return [self initWithCapacity:RACReplaySubjectUnlimitedCapacity];
}

2. 訂閱信號

// 也是在subscribe方法發(fā)生變化，走的是RACReplaySubject中的方法
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];

    RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
        @synchronized (self) {
            // 這里會從valuesReceived里sendNext未被執(zhí)行過value，第三步中會有 valuesReceived 的操作
            for (id value in self.valuesReceived) {
                if (compoundDisposable.disposed) return;
                [subscriber sendNext:(value == RACTupleNil.tupleNil ? nil : value)];
            }

            if (compoundDisposable.disposed) return;

            if (self.hasCompleted) {
                [subscriber sendCompleted];
            } else if (self.hasError) {
                [subscriber sendError:self.error];
            } else {
                RACDisposable *subscriptionDisposable = [super subscribe:subscriber];
                [compoundDisposable addDisposable:subscriptionDisposable];
            }
        }
    }];

    [compoundDisposable addDisposable:schedulingDisposable];

    return compoundDisposable;
}

3. 發(fā)送信號

- (void)sendNext:(id)value {
    @synchronized (self) {
                // 每次將發(fā)送過的value添加到valuesReceived中記錄下來
        [self.valuesReceived addObject:value ?: RACTupleNil.tupleNil];
               // 調(diào)用父類也就是RACSubject的sendNext
        [super sendNext:value];
           // 通過初始化時的capacity值截取valuesReceived的元素(從數(shù)組頭部開始，也就是最先添加進來的value)
        if (self.capacity != RACReplaySubjectUnlimitedCapacity && self.valuesReceived.count > self.capacity) {
            [self.valuesReceived removeObjectsInRange:NSMakeRange(0, self.valuesReceived.count - self.capacity)];
        }
    }
}

為了解釋發(fā)生了什么，進行如下測試

    RACReplaySubject *subject = [RACReplaySubject subject];
    [subject subscribeNext:^(id x) {
        NSLog(@"%@--1",x);
    }];
    [subject subscribeNext:^(id x) {
        NSLog(@"%@--2",x);
    }];
    [subject sendNext:@"send next1"];
    [subject sendNext:@"send next2"];
    [subject subscribeNext:^(id x) {
        NSLog(@"%@--3",x);
    }];

日志：

//1
2017-03-23 22:03:03.722 testdemo[3509:143199] send next1--1
//2
2017-03-23 22:03:03.722 testdemo[3509:143199] send next1--2
//3
2017-03-23 22:03:03.722 testdemo[3509:143199] send next2--1
//4
2017-03-23 22:03:03.723 testdemo[3509:143199] send next2--2
//5
2017-03-23 22:03:03.723 testdemo[3509:143199] send next1--3
//6
2017-03-23 22:03:03.723 testdemo[3509:143199] send next2--3

分析:

  1. [subject sendNext:@"send next1"] 毫無疑問會打印 1 2 兩條日志，同時將value存入了valuesReceived；
  2. [subject sendNext:@"send next2"] 會打印 3 4 兩條日志，也同時將value存入了valuesReceived，到此valuesReceived里就存有兩個value了；
  3. 對subject 進行第三次 subscribeNext 時，因為valuesReceived里是有值的，所以會執(zhí)行兩次subscriber sendNext,也就是打印出日志 5 和 6 。

RACReplaySubject 會將每次sendNext的值記錄到valuesReceived（根據(jù)capacity大小會移除最先添加的元素），之后對RACReplaySubject的訂閱會將valuesReceived里的記錄通過訂閱者來發(fā)送。** 這也就是RACReplaySubject的區(qū)別**

對于RACSignal的介紹到此結(jié)束，我們可以發(fā)現(xiàn)對RACDynamicSignal的每次訂閱都會重復(fù)執(zhí)行didSubscribe，而RACSubject則不會。

在我們平時使用時，很多場景是需要用到 RACDynamicSignal 的。
例如：我們會把數(shù)據(jù)處理、網(wǎng)絡(luò)請求、UI刷新等等一系列操作放在didSubscribe里（這也是RAC的宗旨啊，將平時的各種操作都轉(zhuǎn)化為炫酷的信號的形式，優(yōu)雅的進行傳遞與交互）。但是我們不希望每次對RACSignal的訂閱都重復(fù)執(zhí)行didSubscribe，因為這樣會帶來bug或者性能問題。

這時，我們要做的就是將冷信號轉(zhuǎn)換為熱信號，因為熱信號的重復(fù)訂閱是不會帶來這個問題的。那么如何將冷信號轉(zhuǎn)化為熱信號呢，后面再說吧。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容