RACSignal-冷信號(hào)和熱信號(hào)底層實(shí)現(xiàn)分析

RACSignal冷信號(hào)和熱信號(hào)底層實(shí)現(xiàn)分析前言由于最近在寫關(guān)于RACSignal底層實(shí)現(xiàn)分析的文章，當(dāng)然也逃不了關(guān)于冷熱信號(hào)操作的分析。這篇文章打算分析分析如何從冷信號(hào)轉(zhuǎn)成熱信號(hào)的底層實(shí)現(xiàn)。目錄1.關(guān)于冷信號(hào)和熱信號(hào)的概念2.RACSignal熱信號(hào)3.RACSignal冷信號(hào)4.冷信號(hào)是如何轉(zhuǎn)換成熱信號(hào)的一.關(guān)于冷信號(hào)和熱信號(hào)的概念冷熱信號(hào)的概念是源自于源于.NET框架ReactiveExtensions(RX)中的HotObservable和ColdObservable，HotObservable是主動(dòng)的，盡管你并沒有訂閱事件，但是它會(huì)時(shí)刻推送，就像鼠標(biāo)移動(dòng)；而ColdObservable是被動(dòng)的，只有當(dāng)你訂閱的時(shí)候，它才會(huì)發(fā)布消息。HotObservable可以有多個(gè)訂閱者，是一對(duì)多，集合可以與訂閱者共享信息；而ColdObservable只能一對(duì)一，當(dāng)有不同的訂閱者，消息是重新完整發(fā)送。熱信號(hào)是主動(dòng)的，即使你沒有訂閱事件，它仍然會(huì)時(shí)刻推送。而冷信號(hào)是被動(dòng)的，只有當(dāng)你訂閱的時(shí)候，它才會(huì)發(fā)送消息。熱信號(hào)可以有多個(gè)訂閱者，是一對(duì)多，信號(hào)可以與訂閱者共享信息。而冷信號(hào)只能一對(duì)一，當(dāng)有不同的訂閱者，消息會(huì)從新完整發(fā)送。二.RACSignal熱信號(hào)RACSignal家族中符合熱信號(hào)的特點(diǎn)的信號(hào)有以下幾個(gè)。1.RACSubject@interfaceRACSubject:RACSignal<RACSubscriber>@property(nonatomic,strong,readonly)NSMutableArray*subscribers;@property(nonatomic,strong,readonly)RACCompoundDisposable*disposable;-(void)enumerateSubscribersUsingBlock:(void(^)(id<RACSubscriber>subscriber))block;+(instancetype)subject;@end首先來看看RACSubject的定義。RACSubject是繼承自RACSignal，并且它還遵守RACSubscriber協(xié)議。這就意味著它既能訂閱信號(hào)，也能發(fā)送信號(hào)。在RACSubject里面有一個(gè)NSMutableArray數(shù)組，里面裝著該信號(hào)的所有訂閱者。其次還有一個(gè)RACCompoundDisposable信號(hào)，里面裝著該信號(hào)所有訂閱者的RACDisposable。RACSubject之所以能稱之為熱信號(hào)，那么它肯定是符合上述熱信號(hào)的定義的。讓我們從它的實(shí)現(xiàn)來看看它是如何符合的。-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{NSCParameterAssert(subscriber!=nil);RACCompoundDisposable*disposable=[RACCompoundDisposablecompoundDisposable];subscriber=[[RACPassthroughSubscriberalloc]initWithSubscriber:subscribersignal:selfdisposable:disposable];NSMutableArray*subscribers=self.subscribers;@synchronized(subscribers){[subscribersaddObject:subscriber];}return[RACDisposabledisposableWithBlock:^{@synchronized(subscribers){NSUIntegerindex=[subscribersindexOfObjectWithOptions:NSEnumerationReversepassingTest:^BOOL(id<RACSubscriber>obj,NSUIntegerindex,BOOL*stop){returnobj==subscriber;}];if(index!=NSNotFound)[subscribersremoveObjectAtIndex:index];}}];}上面是RACSubject的實(shí)現(xiàn)，它和RACSignal最大的不同在這兩行NSMutableArray*subscribers=self.subscribers;@synchronized(subscribers){[subscribersaddObject:subscriber];}RACSubject把它的所有訂閱者全部都保存到了NSMutableArray的數(shù)組里。既然保存了所有的訂閱者，那么sendNext，sendError，sendCompleted就需要發(fā)生改變。-(void)sendNext:(id)value{[selfenumerateSubscribersUsingBlock:^(id<RACSubscriber>subscriber){[subscribersendNext:value];}];}-(void)sendError:(NSError*)error{[self.disposabledispose];[selfenumerateSubscribersUsingBlock:^(id<RACSubscriber>subscriber){[subscribersendError:error];}];}-(void)sendCompleted{[self.disposabledispose];[selfenumerateSubscribersUsingBlock:^(id<RACSubscriber>subscriber){[subscribersendCompleted];}];}從源碼可以看到，RACSubject中的sendNext，sendError，sendCompleted都會(huì)執(zhí)行enumerateSubscribersUsingBlock:方法。-(void)enumerateSubscribersUsingBlock:(void(^)(id<RACSubscriber>subscriber))block{NSArray*subscribers;@synchronized(self.subscribers){subscribers=[self.subscriberscopy];}for(id<RACSubscriber>subscriberinsubscribers){block(subscriber);}}enumerateSubscribersUsingBlock:方法會(huì)取出所有RACSubject的訂閱者，依次調(diào)用入?yún)⒌腷lock()方法。關(guān)于RACSubject的訂閱和發(fā)送的流程可以參考第一篇文章，大體一致，其他的不同就是會(huì)依次對(duì)自己的訂閱者發(fā)送信號(hào)。RACSubject就滿足了熱信號(hào)的特點(diǎn)，它即使沒有訂閱者，因?yàn)樽约豪^承了RACSubscriber協(xié)議，所以自己本身就可以發(fā)送信號(hào)。冷信號(hào)只能被訂閱了才能發(fā)送信號(hào)。RACSubject可以有很多訂閱者，它也會(huì)把這些訂閱者都保存到自己的數(shù)組里。RACSubject之后再發(fā)送信號(hào)，訂閱者就如同一起看電視，播放過的節(jié)目就看不到了，發(fā)送過的信號(hào)也接收不到了。接收信號(hào)。而RACSignal發(fā)送信號(hào)，訂閱者接收信號(hào)都只能從頭開始接受，如同看點(diǎn)播節(jié)目，每次看都從頭開始看。2.RACGroupedSignal@interfaceRACGroupedSignal:RACSubject@property(nonatomic,readonly,copy)id<NSCopying>key;+(instancetype)signalWithKey:(id<NSCopying>)key;@end先看看RACGroupedSignal的定義。RACGroupedSignal是在RACsignal這個(gè)方法里面被用到的。-(RACSignal*)groupBy:(id<NSCopying>(^)(idobject))keyBlocktransform:(id(^)(idobject))transformBlock在這個(gè)方法里面，sendNext里面最后里面是由RACGroupedSignal發(fā)送信號(hào)。[groupSubjectsendNext:transformBlock!=NULL?transformBlock(x):x];關(guān)于groupBy的詳細(xì)分析請(qǐng)看這篇文章3.RACBehaviorSubject@interfaceRACBehaviorSubject:RACSubject@property(nonatomic,strong)idcurrentValue;+(instancetype)behaviorSubjectWithDefaultValue:(id)value;@end這個(gè)信號(hào)里面存儲(chǔ)了一個(gè)對(duì)象currentValue，這里存儲(chǔ)著這個(gè)信號(hào)的最新的值。當(dāng)然也可以調(diào)用類方法behaviorSubjectWithDefaultValue+(instancetype)behaviorSubjectWithDefaultValue:(id)value{RACBehaviorSubject*subject=[selfsubject];subject.currentValue=value;returnsubject;}在這個(gè)方法里面存儲(chǔ)默認(rèn)的值，如果RACBehaviorSubject沒有接受到任何值，那么這個(gè)信號(hào)就會(huì)發(fā)送這個(gè)默認(rèn)的值。當(dāng)RACBehaviorSubject被訂閱：-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{RACDisposable*subscriptionDisposable=[supersubscribe:subscriber];RACDisposable*schedulingDisposable=[RACScheduler.subscriptionSchedulerschedule:^{@synchronized(self){[subscribersendNext:self.currentValue];}}];return[RACDisposabledisposableWithBlock:^{[subscriptionDisposabledispose];[schedulingDisposabledispose];}];}sendNext里面會(huì)始終發(fā)送存儲(chǔ)的currentValue值。調(diào)用sendNext會(huì)調(diào)用RACSubject里面的sendNext，也會(huì)依次發(fā)送信號(hào)值給訂閱數(shù)組里面每個(gè)訂閱者。當(dāng)RACBehaviorSubject向訂閱者sendNext的時(shí)候：-(void)sendNext:(id)value{@synchronized(self){self.currentValue=value;[supersendNext:value];}}RACBehaviorSubject會(huì)把發(fā)送的值更新到currentValue里面。下次發(fā)送值就會(huì)發(fā)送最后更新的值。4.RACReplaySubjectconstNSUIntegerRACReplaySubjectUnlimitedCapacity=NSUIntegerMax;@interfaceRACReplaySubject:RACSubject@property(nonatomic,assign,readonly)NSUIntegercapacity;@property(nonatomic,strong,readonly)NSMutableArray*valuesReceived;@property(nonatomic,assign)BOOLhasCompleted;@property(nonatomic,assign)BOOLhasError;@property(nonatomic,strong)NSError*error;+(instancetype)replaySubjectWithCapacity:(NSUInteger)capacity;@endRACReplaySubject中會(huì)存儲(chǔ)RACReplaySubjectUnlimitedCapacity大小的歷史值。+(instancetype)replaySubjectWithCapacity:(NSUInteger)capacity{return[(RACReplaySubject*)[selfalloc]initWithCapacity:capacity];}-(instancetype)init{return[selfinitWithCapacity:RACReplaySubjectUnlimitedCapacity];}-(instancetype)initWithCapacity:(NSUInteger)capacity{self=[superinit];if(self==nil)returnnil;_capacity=capacity;_valuesReceived=(capacity==RACReplaySubjectUnlimitedCapacity?[NSMutableArrayarray]:[NSMutableArrayarrayWithCapacity:capacity]);returnself;}在RACReplaySubject初始化中會(huì)初始化一個(gè)capacity大小的數(shù)組。-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{RACCompoundDisposable*compoundDisposable=[RACCompoundDisposablecompoundDisposable];RACDisposable*schedulingDisposable=[RACScheduler.subscriptionSchedulerschedule:^{@synchronized(self){for(idvalueinself.valuesReceived){if(compoundDisposable.disposed)return;[subscribersendNext:(value==RACTupleNil.tupleNil?nil:value)];}if(compoundDisposable.disposed)return;if(self.hasCompleted){[subscribersendCompleted];}elseif(self.hasError){[subscribersendError:self.error];}else{RACDisposable*subscriptionDisposable=[supersubscribe:subscriber];[compoundDisposableaddDisposable:subscriptionDisposable];}}}];[compoundDisposableaddDisposable:schedulingDisposable];returncompoundDisposable;}當(dāng)RACReplaySubject被訂閱的時(shí)候，會(huì)把valuesReceived數(shù)組里面的值都發(fā)送出去。-(void)sendNext:(id)value{@synchronized(self){[self.valuesReceivedaddObject:value?:RACTupleNil.tupleNil];[supersendNext:value];if(self.capacity!=RACReplaySubjectUnlimitedCapacity&&self.valuesReceived.count>self.capacity){[self.valuesReceivedremoveObjectsInRange:NSMakeRange(0,self.valuesReceived.count-self.capacity)];}}}在sendNext中，valuesReceived會(huì)保存每次接收到的值。調(diào)用super的sendNext，會(huì)依次把值都發(fā)送到每個(gè)訂閱者中。這里還會(huì)判斷數(shù)組里面存儲(chǔ)了多少個(gè)值。如果存儲(chǔ)的值的個(gè)數(shù)大于了capacity，那么要移除掉數(shù)組里面從0開始的前幾個(gè)值，保證數(shù)組里面只裝capacity個(gè)數(shù)的值。RACReplaySubject和RACSubject的區(qū)別在于，RACReplaySubject還會(huì)把歷史的信號(hào)值都存儲(chǔ)起來發(fā)送給訂閱者。這一點(diǎn)，RACReplaySubject更像是RACSingnal和RACSubject的合體版。RACSignal是冷信號(hào)，一旦被訂閱就會(huì)向訂閱者發(fā)送所有的值，這一點(diǎn)RACReplaySubject和RACSignal是一樣的。但是RACReplaySubject又有著RACSubject的特性，會(huì)把所有的值發(fā)送給多個(gè)訂閱者。當(dāng)RACReplaySubject發(fā)送完之前存儲(chǔ)的歷史值之后，之后再發(fā)送信號(hào)的行為就和RACSubject完全一致了。三.RACSignal冷信號(hào)在ReactiveCocoav2.5中除了RACsignal信號(hào)以外，還有一些特殊的冷信號(hào)。1.RACEmptySignal@interfaceRACEmptySignal:RACSignal+(RACSignal*)empty;@end這個(gè)信號(hào)只有一個(gè)empty方法。+(RACSignal*)empty{#ifdefDEBUGreturn[[[selfalloc]init]setNameWithFormat:@"+empty"];#elsestaticidsingleton;staticdispatch_once_tpred;dispatch_once(&pred,^{singleton=[[selfalloc]init];});returnsingleton;#endif}在debug模式下，返回一個(gè)名字叫empty的信號(hào)。在release模式下，返回一個(gè)單例的empty信號(hào)。-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{NSCParameterAssert(subscriber!=nil);return[RACScheduler.subscriptionSchedulerschedule:^{[subscribersendCompleted];}];}RACEmptySignal信號(hào)一旦被訂閱就會(huì)發(fā)送sendCompleted。2.RACReturnSignal@interfaceRACReturnSignal:RACSignal@property(nonatomic,strong,readonly)idvalue;+(RACSignal*)return:(id)value;@endRACReturnSignal信號(hào)的定義也很簡單，直接根據(jù)value的值返回一個(gè)RACSignal。+(RACSignal*)return:(id)value{#ifndefDEBUGif(value==RACUnit.defaultUnit){staticRACReturnSignal*unitSingleton;staticdispatch_once_tunitPred;dispatch_once(&unitPred,^{unitSingleton=[[selfalloc]init];unitSingleton->_value=RACUnit.defaultUnit;});returnunitSingleton;}elseif(value==nil){staticRACReturnSignal*nilSingleton;staticdispatch_once_tnilPred;dispatch_once(&nilPred,^{nilSingleton=[[selfalloc]init];nilSingleton->_value=nil;});returnnilSingleton;}#endifRACReturnSignal*signal=[[selfalloc]init];signal->_value=value;#ifdefDEBUG[signalsetNameWithFormat:@"+return:%@",value];#endifreturnsignal;}在debug模式下直接新建一個(gè)RACReturnSignal信號(hào)里面的值存儲(chǔ)的是入?yún)alue。在release模式下，會(huì)依照value的值是否是空，來新建對(duì)應(yīng)的單例RACReturnSignal。-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{NSCParameterAssert(subscriber!=nil);return[RACScheduler.subscriptionSchedulerschedule:^{[subscribersendNext:self.value];[subscribersendCompleted];}];}RACReturnSignal在被訂閱的時(shí)候，就只會(huì)發(fā)送一個(gè)value值的信號(hào)，發(fā)送完畢之后就sendCompleted。3.RACDynamicSignal這個(gè)信號(hào)是創(chuàng)建RACSignalcreateSignal:的真身。關(guān)于RACDynamicSignal詳細(xì)過程請(qǐng)看第一篇文章。4.RACErrorSignal@interfaceRACErrorSignal:RACSignal@property(nonatomic,strong,readonly)NSError*error;+(RACSignal*)error:(NSError*)error;@endRACErrorSignal信號(hào)里面就存儲(chǔ)了一個(gè)NSError。+(RACSignal*)error:(NSError*)error{RACErrorSignal*signal=[[selfalloc]init];signal->_error=error;#ifdefDEBUG[signalsetNameWithFormat:@"+error:%@",error];#else=@"+error:";#endifreturnsignal;}RACErrorSignal初始化的時(shí)候把外界傳進(jìn)來的Error保存起來。當(dāng)被訂閱的時(shí)候就發(fā)送這個(gè)Error出去。5.RACChannelTerminal@interfaceRACChannelTerminal:RACSignal<RACSubscriber>-(id)init__attribute__((unavailable("InstantiateaRACChannelinstead")));@property(nonatomic,strong,readonly)RACSignal*values;@property(nonatomic,strong,readonly)id<RACSubscriber>otherTerminal;-(id)initWithValues:(RACSignal*)valuesotherTerminal:(id<RACSubscriber>)otherTerminal;@endRACChannelTerminal在RAC日常開發(fā)中，用來雙向綁定的。它和RACSubject一樣，既繼承自RACSignal，同樣又遵守RACSubscriber協(xié)議。雖然具有RACSubject的發(fā)送和接收信號(hào)的特性，但是它依舊是冷信號(hào)，因?yàn)樗鼰o法一對(duì)多，它發(fā)送信號(hào)還是只能一對(duì)一。RACChannelTerminal無法手動(dòng)初始化，需要靠RACChannel去初始化。-(id)init{self=[superinit];if(self==nil)returnnil;RACReplaySubject*leadingSubject=[[RACReplaySubjectreplaySubjectWithCapacity:0]setNameWithFormat:@"leadingSubject"];RACReplaySubject*followingSubject=[[RACReplaySubjectreplaySubjectWithCapacity:1]setNameWithFormat:@"followingSubject"];[[leadingSubjectignoreValues]subscribe:followingSubject];[[followingSubjectignoreValues]subscribe:leadingSubject];_leadingTerminal=[[[RACChannelTerminalalloc]initWithValues:leadingSubjectotherTerminal:followingSubject]setNameWithFormat:@"leadingTerminal"];_followingTerminal=[[[RACChannelTerminalalloc]initWithValues:followingSubjectotherTerminal:leadingSubject]setNameWithFormat:@"followingTerminal"];returnself;}在RACChannel的初始化中會(huì)調(diào)用RACChannelTerminal的initWithValues:方法，這里的入?yún)⒍际荝ACReplaySubject類型的。所以訂閱RACChannelTerminal過程的時(shí)候：-(RACDisposable*)subscribe:(id<RACSubscriber>)subscriber{return[self.valuessubscribe:subscriber];}self.values其實(shí)就是一個(gè)RACReplaySubject，就相當(dāng)于訂閱RACReplaySubject。訂閱過程同上面RACReplaySubject的訂閱過程。-(void)sendNext:(id)value{[self.otherTerminalsendNext:value];}-(void)sendError:(NSError*)error{[self.otherTerminalsendError:error];}-(void)sendCompleted{[self.otherTerminalsendCompleted];}self.otherTerminal也是RACReplaySubject類型的，RACChannelTerminal管道兩邊都是RACReplaySubject類型的信號(hào)。當(dāng)RACChannelTerminal開始sendNext，sendError，sendCompleted是調(diào)用的管道另外一個(gè)的RACReplaySubject進(jìn)行這些對(duì)應(yīng)的操作的。平時(shí)使用RACChannelTerminal的地方在View和ViewModel的雙向綁定上面。例如在登錄界面，輸入密碼文本框TextField和ViewModel的Password雙向綁定RACChannelTerminal*passwordTerminal=[_passwordTextFieldrac_newTextChannel];RACChannelTerminal*viewModelPasswordTerminal=RACChannelTo(_viewModel,rd);[viewModelPasswordTerminalsubscribe:passwordTerminal];[passwordTerminalsubscribe:viewModelPasswordTerminal];雙向綁定的兩個(gè)信號(hào)都會(huì)因?yàn)閷?duì)方的改變而收到新的信號(hào)。至此所有的RACSignal的分類就都理順了，按照冷信號(hào)和熱信號(hào)的分類也分好了。根據(jù)RACSignal訂閱和發(fā)送信號(hào)的流程，我們可以知道，每訂閱一次冷信號(hào)RACSignal，就會(huì)執(zhí)行一次didSubscribe閉包。這個(gè)時(shí)候就是可能出現(xiàn)問題的地方。如果RACSignal是被用于網(wǎng)絡(luò)請(qǐng)求，那么在didSubscribe閉包里面會(huì)被重復(fù)的請(qǐng)求。上面文中提到了信號(hào)被訂閱了6次，網(wǎng)絡(luò)請(qǐng)求也會(huì)請(qǐng)求6次。這并不是我們想要的。網(wǎng)絡(luò)請(qǐng)求只需要請(qǐng)求1次。如何做到信號(hào)只執(zhí)行一次didSubscribe閉包，最重要的一點(diǎn)是RACSignal冷信號(hào)只能被訂閱一次。由于冷信號(hào)只能一對(duì)一，那么想一對(duì)多就只能交給熱信號(hào)去處理了。這時(shí)候就需要把冷信號(hào)轉(zhuǎn)換成熱信號(hào)。在ReactiveCocoav2.5中，冷信號(hào)轉(zhuǎn)換成熱信號(hào)需要用到RACMulticastConnection這個(gè)類。@interfaceRACMulticastConnection:NSObject@property(nonatomic,strong,readonly)RACSignal*signal;-(RACDisposable*)connect;-(RACSignal*)autoconnect;@end@interfaceRACMulticastConnection(){RACSubject*_signal;int32_tvolatile_hasConnected;}@property(nonatomic,readonly,strong)RACSignal*sourceSignal;@property(strong)RACSerialDisposable*serialDisposable;@end看看RACMulticastConnection類的定義。最主要的是保存了兩個(gè)信號(hào)，一個(gè)是RACSubject，一個(gè)是sourceSignal(RACSignal類型)。在.h中暴露給外面的是RACSignal，在.m中實(shí)際使用的是RACSubject?？此亩x就能猜到接下去它會(huì)做什么：用sourceSignal去發(fā)送信號(hào)，內(nèi)部再用RACSubject去訂閱sourceSignal，然后RACSubject會(huì)把sourceSignal的信號(hào)值依次發(fā)給它的訂閱者們。用一個(gè)不恰當(dāng)?shù)谋扔鱽硇稳軷ACMulticastConnection，它就像上圖中心的那個(gè)“地球”，“地球”就是訂閱了sourceSignal的RACSubject，RACSubject把值發(fā)送給各個(gè)“連接”者(訂閱者)。sourceSignal只有內(nèi)部的RACSubject一個(gè)訂閱者，所以就完成了我們只想執(zhí)行didSubscribe閉包一次，但是能把值發(fā)送給各個(gè)訂閱者的愿望。在看看RACMulticastConnection的初始化-(id)initWithSourceSignal:(RACSignal*)sourcesubject:(RACSubject*)subject{NSCParameterAssert(source!=nil);NSCParameterAssert(subject!=nil);self=[superinit];if(self==nil)returnnil;_sourceSignal=source;_serialDisposable=[[RACSerialDisposablealloc]init];_signal=subject;returnself;}初始化方法就是把外界傳進(jìn)來的RACSignal保存成sourceSignal，把外界傳進(jìn)來的RACSubject保存成自己的signal屬性。RACMulticastConnection有兩個(gè)連接方法。-(RACDisposable*)connect{BOOLshouldConnect=OSAtomicCompareAndSwap32Barrier(0,1,&_hasConnected);if(shouldConnect){self.serialDisposable.disposable=[self.sourceSignalsubscribe:_signal];}returnself.serialDisposable;}這里出現(xiàn)了一個(gè)不多見的函數(shù)OSAtomicCompareAndSwap32Barrier，它是原子運(yùn)算的操作符，主要用于Compareandswap，原型如下：boolOSAtomicCompareAndSwap32Barrier(int32_t__oldValue,int32_t__newValue,volatileint32_t*__theValue);關(guān)鍵字volatile只確保每次獲取volatile變量時(shí)都是從內(nèi)存加載變量，而不是使用寄存器里面的值，但是它不保證代碼訪問變量是正確的。如果用偽代碼去實(shí)現(xiàn)這個(gè)函數(shù):f(*__theValue==__oldValue){*__theValue=__newValue;return1;}else{return0;}如果_hasConnected為0，意味著沒有連接，OSAtomicCompareAndSwap32Barrier返回1，shouldConnect就應(yīng)該連接。如果_hasConnected為1，意味著已經(jīng)連接過了，OSAtomicCompareAndSwap32Barrier返回0，shouldConnect不會(huì)再次連接。所謂連接的過程就是RACMulticastConnection內(nèi)部用RACSubject訂閱self.sourceSignal。sourceSignal是RACSignal，會(huì)把訂閱者RACSubject保存到RACPassthroughSubscriber中，sendNext的時(shí)候就會(huì)調(diào)用RACSubjectsendNext，這時(shí)就會(huì)把sourceSignal的信號(hào)都發(fā)送給各個(gè)訂閱者了。-(RACSignal*)autoconnect{__blockvolatileint32_tsubscriberCount=0;return[[RACSignalcreateSignal:^(id<RACSubscriber>subscriber){OSAtomicIncrement32Barrier(&subscriberCount);RACDisposable*subscriptionDisposable=[self.signalsubscribe:subscriber];RACDisposable*connectionDisposable=[selfconnect];return[RACDisposabledisposableWithBlock:^{[subscriptionDisposabledispose];if(OSAtomicDecrement32Barrier(&subscriberCount)==0){[connectionDisposabledispose];}}];}]setNameWithFormat:@"[%@]-autoconnect",];}OSAtomicIncrement32Barrier和OSAtomicDecrement32Barrier也是原子運(yùn)算的操作符，分別是+1和-1操作。在autoconnect為了保證線程安全，用到了一個(gè)subscriberCount的類似信號(hào)量的volatile變量，保證第一個(gè)訂閱者能連接上。返回的新的信號(hào)的訂閱者訂閱RACSubject，RACSubject也會(huì)去訂閱內(nèi)部的sourceSignal。把冷信號(hào)轉(zhuǎn)換成熱信號(hào)用以下5種方式，5種方法都會(huì)用到RACMulticastConnection。接下來一一分析它們的具體實(shí)現(xiàn)。1.multicast:-(RACMulticastConnection*)multicast:(RACSubject*)subject{[subjectsetNameWithFormat:@"[%@]-multicast:%@",,];RACMulticastConnection*connection=[[RACMulticastConnectionalloc]initWithSourceSignal:selfsubject:subject];returnconnection;}multicast：的操作就是初始化一個(gè)RACMulticastConnection對(duì)象，SourceSignal是self，內(nèi)部的RACSubject是入?yún)ubject。RACMulticastConnection*connection=[signalmulticast:[RACSubjectsubject]];[connection.signalsubscribeNext:^(idx){NSLog(@"%@",x);}];[connectionconnect];調(diào)用multicast:把冷信號(hào)轉(zhuǎn)換成熱信號(hào)有一個(gè)點(diǎn)不方便的是，需要自己手動(dòng)connect。注意轉(zhuǎn)換完之后的熱信號(hào)在RACMulticastConnection的signal屬性中，所以需要訂閱的是connection.signal。2.publish-(RACMulticastConnection*)publish{RACSubject*subject=[[RACSubjectsubject]setNameWithFormat:@"[%@]-publish",me];RACMulticastConnection*connection=[selfmulticast:subject];returnconnection;}publish方法只不過是去調(diào)用了multicast:方法，publish內(nèi)部會(huì)新建好一個(gè)RACSubject，并把它當(dāng)成入?yún)鬟f給RACMulticastConnection。RACMulticastConnection*connection=[signalpublish];[connection.signalsubscribeNext:^(idx){NSLog(@"%@",x);}];[connectionconnect];同樣publish方法也需要手動(dòng)的調(diào)用connect方法。3.replay-(RACSignal*)replay{RACReplaySubject*subject=[[RACReplaySubjectsubject]setNameWithFormat:@"[%@]-replay",];RACMulticastConnection*connection=[selfmulticast:subject];[connectionconnect];