iOS即时通讯之CocoaAsyncSocket源码解析二

原文

前言

本文承接上文:iOS即时通讯之CocoaAsyncSocket源码解析一

上文我们提到了GCDAsyncSocket的初始化，以及最终connect之前的准备工作，包括一些错误检查；本机地址创建以及socket创建；服务端地址的创建；还有一些本机socket可选项的配置，例如禁止网络出错导致进程关闭的信号等。

 上文讲到了本文方法八--创建Socket，其中有这么一行代码：

//和connectInterface绑定
if (![self bindSocket:socketFD toInterface:connectInterface error:errPtr])
{
    //绑定失败，直接关闭返回
    [self closeSocket:socketFD];

    return SOCKET_NULL;
}

我们去用之前创建的本机地址去做socket绑定，接着会调用到如下方法中：

本文方法九--给Socket绑定本机地址

 1 //绑定一个Socket的本地地址
 2 - (BOOL)bindSocket:(int)socketFD toInterface:(NSData *)connectInterface error:(NSError **)errPtr
 3 {
 4     // Bind the socket to the desired interface (if needed)
 5     //无接口就不绑定，connect会自动绑定到一个不冲突的端口上去。
 6     if (connectInterface)
 7     {
 8         LogVerbose(@"Binding socket...");
 9 
10         //判断当前地址的Port是不是大于0
11         if ([[self class] portFromAddress:connectInterface] > 0)
12         {
13             // Since we're going to be binding to a specific port,
14             // we should turn on reuseaddr to allow us to override sockets in time_wait.
15 
16             int reuseOn = 1;
17 
18 
19             //设置调用close(socket)后,仍可继续重用该socket。调用close(socket)一般不会立即关闭socket，而经历TIME_WAIT的过程。
20             setsockopt(socketFD, SOL_SOCKET, SO_REUSEADDR, &reuseOn, sizeof(reuseOn));
21         }
22 
23         //拿到地址
24         const struct sockaddr *interfaceAddr = (const struct sockaddr *)[connectInterface bytes];
25         //绑定这个地址
26         int result = bind(socketFD, interfaceAddr, (socklen_t)[connectInterface length]);
27 
28         //绑定出错，返回NO
29         if (result != 0)
30         {
31             if (errPtr)
32                 *errPtr = [self errnoErrorWithReason:@"Error in bind() function"];
33 
34             return NO;
35         }
36     }
37 
38     //成功
39     return YES;
40 }

这个方法也非常简单，如果没有connectInterface则直接返回YES，当socket进行连接的时候，会自动绑定一个端口，进行连接。
如果有值，则我们开始绑定到我们一开始指定的地址上。
这里调用了两个和scoket相关的函数：
第一个是我们之前提到的配置scoket参数的函数：

setsockopt(socketFD, SOL_SOCKET, SO_REUSEADDR, &reuseOn, sizeof(reuseOn));

这里调用这个函数的主要目的是为了调用close的时候，不立即去关闭socket连接，而是经历一个TIME_WAIT过程。在这个过程中，socket是可以被复用的。我们注意到之前的connect流程并没有看到复用socket的代码。注意，我们现在走的连接流程是客户端的流程，等我们讲到服务端accept进行连接的时候，我们就能看到这个复用的作用了。

第二个是bind函数

int result = bind(socketFD, interfaceAddr, (socklen_t)[connectInterface length]);

这个函数倒是很简单，就3个参数，socket、需要绑定的地址、地址大小。这样就把socket和这个地址（其实就是端口）捆绑在一起了。

这样我们就做完了最终连接前所有准备工作，本机socket有了，服务端的地址也有了。接着我们就可以开始进行最终连接了：

 本文方法十 -- 建立连接的最终方法

 1 //连接最终方法 3 finnal。。。
 2 - (void)connectSocket:(int)socketFD address:(NSData *)address stateIndex:(int)aStateIndex
 3 {
 4     // If there already is a socket connected, we close socketFD and return
 5     //已连接，关闭连接返回
 6     if (self.isConnected)
 7     {
 8         [self closeSocket:socketFD];
 9         return;
10     }
11 
12     // Start the connection process in a background queue
13     //开始连接过程，在后台queue中
14     __weak GCDAsyncSocket *weakSelf = self;
15 
16     //获取到全局Queue
17     dispatch_queue_t globalConcurrentQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
18     //新线程
19     dispatch_async(globalConcurrentQueue, ^{
20 #pragma clang diagnostic push
21 #pragma clang diagnostic warning "-Wimplicit-retain-self"
22         //调用connect方法，该函数阻塞线程，所以要异步新线程
23         //客户端向特定网络地址的服务器发送连接请求，连接成功返回0，失败返回 -1。
24         int result = connect(socketFD, (const struct sockaddr *)[address bytes], (socklen_t)[address length]);
25 
26         //老样子，安全判断
27         __strong GCDAsyncSocket *strongSelf = weakSelf;
28         if (strongSelf == nil) return_from_block;
29 
30         //在socketQueue中，开辟线程
31         dispatch_async(strongSelf->socketQueue, ^{ @autoreleasepool {
32             //如果状态为已经连接，关闭连接返回
33             if (strongSelf.isConnected)
34             {
35                 [strongSelf closeSocket:socketFD];
36                 return_from_block;
37             }
38 
39             //说明连接成功
40             if (result == 0)
41             {
42                 //关闭掉另一个没用的socket
43                 [self closeUnusedSocket:socketFD];
44                 //调用didConnect，生成stream，改变状态等等！
45                 [strongSelf didConnect:aStateIndex];
46             }
47             //连接失败
48             else
49             {
50                 //关闭当前socket
51                 [strongSelf closeSocket:socketFD];
52 
53                 // If there are no more sockets trying to connect, we inform the error to the delegate
54                 //返回连接错误的error
55                 if (strongSelf.socket4FD == SOCKET_NULL && strongSelf.socket6FD == SOCKET_NULL)
56                 {
57                     NSError *error = [strongSelf errnoErrorWithReason:@"Error in connect() function"];
58                     [strongSelf didNotConnect:aStateIndex error:error];
59                 }
60             }
61         }});
62 
63 #pragma clang diagnostic pop
64     });
65     //输出正在连接中
66     LogVerbose(@"Connecting...");
67 }

这个方法主要就是做了一件事，调用下面一个函数进行连接：

int result = connect(socketFD, (const struct sockaddr *)[address bytes], (socklen_t)[address length]);

这里需要注意的是这个函数是阻塞，直到结果返回之前，线程会一直停在这行。所以这里用的是全局并发队列，开辟了一个新的线程进行连接，在得到结果之后，又调回socketQueue中进行后续操作。

如果result为0，说明连接成功，我们会关闭掉另外一个没有用到的socket（如果有的话）。然后调用另外一个方法做一些连接成功的初始化操作。
否则连接失败，我们会关闭socket，填充错误并且返回。

我们接着来看看连接成功后，初始化的方法：

本文方法十一 -- 连接成功后的初始化

  1 //连接成功后调用，设置一些连接成功的状态
  2 - (void)didConnect:(int)aStateIndex
  3 {
  4     LogTrace();
  5 
  6     NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
  7 
  8     //状态不同
  9     if (aStateIndex != stateIndex)
 10     {
 11         LogInfo(@"Ignoring didConnect, already disconnected");
 12 
 13         // The connect operation has been cancelled.
 14         // That is, socket was disconnected, or connection has already timed out.
 15         return;
 16     }
 17 
 18     //kConnected合并到当前flag中
 19     flags |= kConnected;
 20     //停止连接超时
 21     [self endConnectTimeout];
 22 
 23     #if TARGET_OS_IPHONE
 24     // The endConnectTimeout method executed above incremented the stateIndex.
 25     //上面的endConnectTimeout,会导致stateIndex增加，所以需要重新赋值
 26     aStateIndex = stateIndex;
 27     #endif
 28 
 29     // Setup read/write streams (as workaround for specific shortcomings in the iOS platform)
 30     // 
 31     // Note:
 32     // There may be configuration options that must be set by the delegate before opening the streams.
 33     //打开stream之前必须用相关配置设置代理
 34     // The primary example is the kCFStreamNetworkServiceTypeVoIP flag, which only works on an unopened stream.
 35     //主要的例子是kCFStreamNetworkServiceTypeVoIP标记，只能工作在未打开的stream中？
 36     // 
 37     // Thus we wait until after the socket:didConnectToHost:port: delegate method has completed.
 38     //所以我们要等待，连接完成的代理调用完
 39     // This gives the delegate time to properly configure the streams if needed.
 40     //这些给了代理时间，去正确的配置Stream，如果是必要的话
 41 
 42     //创建个Block来初始化Stream
 43     dispatch_block_t SetupStreamsPart1 = ^{
 44 
 45         NSLog(@"hello~");
 46         #if TARGET_OS_IPHONE
 47         //创建读写stream失败，则关闭并报对应错误
 48         if (![self createReadAndWriteStream])
 49         {
 50             [self closeWithError:[self otherError:@"Error creating CFStreams"]];
 51             return;
 52         }
 53 
 54         //参数是给NO的，就是有可读bytes的时候，不会调用回调函数
 55         if (![self registerForStreamCallbacksIncludingReadWrite:NO])
 56         {
 57             [self closeWithError:[self otherError:@"Error in CFStreamSetClient"]];
 58             return;
 59         }
 60 
 61         #endif
 62     };
 63     //part2设置stream
 64     dispatch_block_t SetupStreamsPart2 = ^{
 65         #if TARGET_OS_IPHONE
 66         //状态不一样直接返回
 67         if (aStateIndex != stateIndex)
 68         {
 69             // The socket has been disconnected.
 70             return;
 71         }
 72         //如果加到runloop上失败
 73         if (![self addStreamsToRunLoop])
 74         {
 75             //错误返回
 76             [self closeWithError:[self otherError:@"Error in CFStreamScheduleWithRunLoop"]];
 77             return;
 78         }
 79 
 80         //读写stream open
 81         if (![self openStreams])
 82         {
 83             //开启错误返回
 84             [self closeWithError:[self otherError:@"Error creating CFStreams"]];
 85             return;
 86         }
 87 
 88         #endif
 89     };
 90 
 91     // Notify delegate
 92     //通知代理
 93     //拿到server端的host port
 94     NSString *host = [self connectedHost];
 95     uint16_t port = [self connectedPort];
 96     //拿到unix域的 url
 97     NSURL *url = [self connectedUrl];
 98     //拿到代理
 99     __strong id theDelegate = delegate;
100 
101     //代理队列 和 Host不为nil 且响应didConnectToHost代理方法
102     if (delegateQueue && host != nil && [theDelegate respondsToSelector:@selector(socket:didConnectToHost:port:)])
103     {
104         //调用初始化stream1
105         SetupStreamsPart1();
106 
107         dispatch_async(delegateQueue, ^{ @autoreleasepool {
108 
109             //到代理队列调用连接成功的代理方法
110             [theDelegate socket:self didConnectToHost:host port:port];
111 
112             //然后回到socketQueue中去执行初始化stream2
113             dispatch_async(socketQueue, ^{ @autoreleasepool {
114 
115                 SetupStreamsPart2();
116             }});
117         }});
118     }
119     //这个是unix domain 请求回调
120     else if (delegateQueue && url != nil && [theDelegate respondsToSelector:@selector(socket:didConnectToUrl:)])
121     {
122         SetupStreamsPart1();
123 
124         dispatch_async(delegateQueue, ^{ @autoreleasepool {
125 
126             [theDelegate socket:self didConnectToUrl:url];
127 
128             dispatch_async(socketQueue, ^{ @autoreleasepool {
129 
130                 SetupStreamsPart2();
131             }});
132         }});
133     }
134     //否则只初始化stream
135     else
136     {
137         SetupStreamsPart1();
138         SetupStreamsPart2();
139     }
140 
141     // Get the connected socket
142 
143     int socketFD = (socket4FD != SOCKET_NULL) ? socket4FD : (socket6FD != SOCKET_NULL) ? socket6FD : socketUN;
144 
145     //fcntl,功能描述：根据文件描述词来操作文件的特性。http://blog.csdn.net/pbymw8iwm/article/details/7974789
146     // Enable non-blocking IO on the socket
147     //使socket支持非阻塞IO
148     int result = fcntl(socketFD, F_SETFL, O_NONBLOCK);
149     if (result == -1)
150     {
151         //失败 ，报错
152         NSString *errMsg = @"Error enabling non-blocking IO on socket (fcntl)";
153         [self closeWithError:[self otherError:errMsg]];
154 
155         return;
156     }
157 
158     // Setup our read/write sources
159     //初始化读写source
160     [self setupReadAndWriteSourcesForNewlyConnectedSocket:socketFD];
161 
162     // Dequeue any pending read/write requests
163     //开始下一个任务
164     [self maybeDequeueRead];
165     [self maybeDequeueWrite];
166 }

这个方法很长一大串，其实做的东西也很简单，主要做了下面几件事：

  1. 把当前状态flags加上已连接，并且关闭掉我们一开始连接开启的，连接超时的定时器。
  2. 初始化了两个Block：SetupStreamsPart1、SetupStreamsPart2，这两个Block做的事都和读写流有关。SetupStreamsPart1用来创建读写流，并且注册回调。另一个SetupStreamsPart2用来把流添加到当前线程的runloop上，并且打开流。
  3. 判断是否有代理queue、host或者url这些参数是否为空、是否代理响应didConnectToHost或didConnectToUrl代理，这两种分别对应了普通socket连接和unix domin socket连接。如果实现了对应的代理，则调用连接成功的代理。

  4. 在调用代理的同时，调用了我们之前初始化的两个读写流相关的Block。这里值得说下的是这两个Block和代理之间的调用顺序：

    • 先执行SetupStreamsPart1后执行SetupStreamsPart2，没什么好说的，问题是代理的执行时间，想想如果我们放在SetupStreamsPart2后面是不是会导致个问题，就是用户收到消息了，但是连接成功的代理还没有被调用，这显然是不合理的。所以我们的调用顺序是SetupStreamsPart1->代理->SetupStreamsPart2

      所以出现了如下代码：

    •  1 //调用初始化stream1
 2    SetupStreamsPart1();
 3 
 4    dispatch_async(delegateQueue, ^{ @autoreleasepool {
 5 
 6        //到代理队列调用连接成功的代理方法
 7        [theDelegate socket:self didConnectToHost:host port:port];
 8 
 9        //然后回到socketQueue中去执行初始化stream2
10        dispatch_async(socketQueue, ^{ @autoreleasepool {
11 
12            SetupStreamsPart2();
13        }});
14    }});

原因是为了线程安全和socket相关的操作必须在socketQueue中进行。而代理必须在我们设置的代理queue中被回调。

5、拿到当前的本机socket，调用如下函数：

int result = fcntl(socketFD, F_SETFL, O_NONBLOCK);

简单来说，这个函数类似我们之前提到的一个函数setsockopt()，都是给socket设置一些参数，以实现一些功能。而这个函数，能实现的功能更多。大家可以看看这篇文章参考参考：fcntl函数详解

而在这里，就是为了把socket的IO模式设置为非阻塞。很多小伙伴又要疑惑什么是非阻塞了，先别急，关于这个我们下文会详细的来谈。

6、我们初始化了读写source（很重要，所有的消息都是由这个source来触发的，我们之后会详细分析这个方法）。

7、我们做完了stream和source的初始化处理，则开始做一次读写任务（这两个方法暂时不讲，会放到之后的Read和Write篇中去讲）。

我们接着来讲讲这个方法中对其他方法的调用，按照顺序来，先从第2条，两个Block中对stream的处理开始。和stream相关的函数一共有6个：

Stream相关方法一 -- 创建读写stream 

 1 //创建读写stream
 2 - (BOOL)createReadAndWriteStream
 3 {
 4     LogTrace();
 5 
 6     NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
 7 
 8     //如果有一个有值，就返回
 9     if (readStream || writeStream)
10     {
11         // Streams already created
12         return YES;
13     }
14     //拿到socket，首选是socket4FD，其次socket6FD，都没有才是socketUN，socketUN应该是Unix的socket结构体
15     int socketFD = (socket4FD != SOCKET_NULL) ? socket4FD : (socket6FD != SOCKET_NULL) ? socket6FD : socketUN;
16 
17     //如果都为空，返回NO
18     if (socketFD == SOCKET_NULL)
19     {
20         // Cannot create streams without a file descriptor
21         return NO;
22     }
23 
24     //如果非连接，返回NO
25     if (![self isConnected])
26     {
27         // Cannot create streams until file descriptor is connected
28         return NO;
29     }
30 
31     LogVerbose(@"Creating read and write stream...");
32 
33 #pragma mark - 绑定Socket和CFStream
34     //下面的接口用于创建一对 socket stream，一个用于读取，一个用于写入：
35     CFStreamCreatePairWithSocket(NULL, (CFSocketNativeHandle)socketFD, &readStream, &writeStream);
36 
37     // The kCFStreamPropertyShouldCloseNativeSocket property should be false by default (for our case).
38     // But let's not take any chances.
39 
40 
41 
42     //读写stream都设置成不会随着绑定的socket一起close,release。 kCFBooleanFalse不一起，kCFBooleanTrue一起
43     if (readStream)
44         CFReadStreamSetProperty(readStream, kCFStreamPropertyShouldCloseNativeSocket, kCFBooleanFalse);
45     if (writeStream)
46         CFWriteStreamSetProperty(writeStream, kCFStreamPropertyShouldCloseNativeSocket, kCFBooleanFalse);
47 
48     //如果有一个为空
49     if ((readStream == NULL) || (writeStream == NULL))
50     {
51         LogWarn(@"Unable to create read and write stream...");
52 
53         //关闭对应的stream
54         if (readStream)
55         {
56             CFReadStreamClose(readStream);
57             CFRelease(readStream);
58             readStream = NULL;
59         }
60         if (writeStream)
61         {
62             CFWriteStreamClose(writeStream);
63             CFRelease(writeStream);
64             writeStream = NULL;
65         }
66         //返回创建失败
67         return NO;
68     }
69     //创建成功
70     return YES;
71 }

这个方法基本上很简单，就是关于两个stream函数的调用：

1、创建stream的函数：

  1. CFStreamCreatePairWithSocket(NULL, (CFSocketNativeHandle)socketFD, &readStream, &writeStream);

这个函数创建了一对读写stream，并且把stream与这个scoket做了绑定。

2、设置stream属性：

  1. CFReadStreamSetProperty(readStream, kCFStreamPropertyShouldCloseNativeSocket, kCFBooleanFalse);
CFWriteStreamSetProperty(writeStream, kCFStreamPropertyShouldCloseNativeSocket, kCFBooleanFalse);

    这个函数可以给stream设置一个属性，这里是设置stream不会随着socket的生命周期（close,release）而变化。

接着调用了registerForStreamCallbacksIncludingReadWrite来给stream注册读写回调。

 Stream相关方法二 -- 读写回调的注册：

//注册Stream的回调
- (BOOL)registerForStreamCallbacksIncludingReadWrite:(BOOL)includeReadWrite
{
    LogVerbose(@"%@ %@", THIS_METHOD, (includeReadWrite ? @"YES" : @"NO"));

    NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
    //判断读写stream是不是都为空
    NSAssert((readStream != NULL && writeStream != NULL), @"Read/Write stream is null");

    //客户端stream上下文对象
    streamContext.version = 0;
    streamContext.info = (__bridge void *)(self);
    streamContext.retain = nil;
    streamContext.release = nil;
    streamContext.copyDescription = nil;

//    The open has completed successfully.
//    The stream has bytes to be read.
//    The stream can accept bytes for writing.
//        An error has occurred on the stream.
//        The end of the stream has been reached.

    //设置一个CF的flag  两种，一种是错误发生的时候，一种是stream事件结束
    CFOptionFlags readStreamEvents = kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered ;
    //如果包含读写
    if (includeReadWrite)
        //仍然有Bytes要读的时候     The stream has bytes to be read.
        readStreamEvents |= kCFStreamEventHasBytesAvailable;

    //给读stream设置客户端，会在之前设置的那些标记下回调函数 CFReadStreamCallback。设置失败的话直接返回NO
    if (!CFReadStreamSetClient(readStream, readStreamEvents, &CFReadStreamCallback, &streamContext))
    {
        return NO;
    }

    //写的flag,也一样
    CFOptionFlags writeStreamEvents = kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered;
    if (includeReadWrite)
        writeStreamEvents |= kCFStreamEventCanAcceptBytes;

    if (!CFWriteStreamSetClient(writeStream, writeStreamEvents, &CFWriteStreamCallback, &streamContext))
    {
        return NO;
    }
    //走到最后说明读写都设置回调成功，返回YES
    return YES;
}

相信用过CFStream的朋友，应该会觉得很简单，这个方法就是调用了一些CFStream相关函数，其中最主要的这个设置读写回调函数：

Boolean CFReadStreamSetClient(CFReadStreamRef stream, CFOptionFlags streamEvents, CFReadStreamClientCallBack clientCB, CFStreamClientContext *clientContext);
Boolean CFWriteStreamSetClient(CFWriteStreamRef stream, CFOptionFlags streamEvents, CFWriteStreamClientCallBack clientCB, CFStreamClientContext *clientContext);

这个函数共4个参数：
第1个为我们需要设置的stream；
第2个为需要监听的事件选项，包括以下事件：

typedef CF_OPTIONS(CFOptionFlags, CFStreamEventType) {
    kCFStreamEventNone = 0,  //没有事件发生
    kCFStreamEventOpenCompleted = 1,  //成功打开流
    kCFStreamEventHasBytesAvailable = 2, //流中有数据可读
    kCFStreamEventCanAcceptBytes = 4,  //流中可以接受数据去写
    kCFStreamEventErrorOccurred = 8,  //流发生错误
    kCFStreamEventEndEncountered = 16  //到达流的结尾
};

其中具体用法，大家可以自行去试试，这里作者只监听了了两种事件kCFStreamEventErrorOccurred和kCFStreamEventEndEncountered，再根据传过来的参数去决定是否监听kCFStreamEventCanAcceptBytes：

//如果包含读写
if (includeReadWrite)
   //仍然有Bytes要读的时候     The stream has bytes to be read.
     readStreamEvents |= kCFStreamEventHasBytesAvailable;

而这里我们传过来的参数为NO，导致它并不监听可读数据。显然，我们正常的连接，当有消息发送过来，并不是由stream回调来触发的。这个框架中，如果是TLS传输的socket是用stream来触发的，这个我们后续文章会讲到。

那么有数据的时候，到底是什么来触发我们的读写呢，答案就是读写source，我们接下来就会去创建初始化它。

这里绑定了两个函数，分别对应读和写的回调，分别为：

//读的回调
static void CFReadStreamCallback (CFReadStreamRef stream, CFStreamEventType type, void *pInfo)
//写的回调
static void CFWriteStreamCallback (CFWriteStreamRef stream, CFStreamEventType type, void *pInfo)

关于这两个函数，同样这里暂时不做讨论，等后续文章再来分析。

还有一点需要说一下的是streamContext这个属性，它是一个结构体，包含流的上下文信息，其结构如下：

typedef struct {
    CFIndex version;
    void *info;
    void *(*retain)(void *info);
    void (*release)(void *info);
    CFStringRef (*copyDescription)(void *info);
} CFStreamClientContext

这个流的上下文中info指针，其实就是前面所对应的读写回调函数中的pInfo指针，每次回调都会传过去。其它的version就是流的版本标识，之外的3个都需要的是一个函数指针，对应我们传递的pInfo的持有以及释放还有复制的描述信息，这里我们都赋值给nil。

接着我们来到流处理的第三步：addStreamsToRunLoop-添加到runloop上。

Stream相关方法三 -- 加到当前线程的runloop

//把stream添加到runloop上
- (BOOL)addStreamsToRunLoop
{
    LogTrace();

    NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
    NSAssert((readStream != NULL && writeStream != NULL), @"Read/Write stream is null");

    //判断flag里是否包含kAddedStreamsToRunLoop，没添加过则添加。
    if (!(flags & kAddedStreamsToRunLoop))
    {
        LogVerbose(@"Adding streams to runloop...");


        [[self class] startCFStreamThreadIfNeeded];
        //在开启的线程中去执行，阻塞式的
        [[self class] performSelector:@selector(scheduleCFStreams:)
                             onThread:cfstreamThread
                           withObject:self
                        waitUntilDone:YES];

        //添加标识
        flags |= kAddedStreamsToRunLoop;
    }

    return YES;
}

这里方法做了两件事：

1、开启了一条用于CFStream读写回调的常驻线程，其中调用了好几个函数：

  1. + (void)startCFStreamThreadIfNeeded；
+ (void)cfstreamThread；

    在这两个函数中，添加了一个runloop，并且绑定了一个定时器事件，让它run起来，使得线程常驻。大家可以结合着github中demo的注释，自行查看这几个方法。如果有任何疑问可以看看楼主这篇文章：基于runloop的线程保活、销毁与通信，或者本文下评论，会一一解答。

2、在这个常驻线程中去调用注册方法：

  1.  1 //注册CFStream
 2 + (void)scheduleCFStreams:(GCDAsyncSocket *)asyncSocket
 3 {
 4  LogTrace();
 5 
 6  //断言当前线程是cfstreamThread，不是则报错
 7  NSAssert([NSThread currentThread] == cfstreamThread, @"Invoked on wrong thread");
 8 
 9  //获取到runloop
10  CFRunLoopRef runLoop = CFRunLoopGetCurrent();
11  //如果有readStream
12  if (asyncSocket->readStream)
13      //注册readStream在runloop的kCFRunLoopDefaultMode上
14      CFReadStreamScheduleWithRunLoop(asyncSocket->readStream, runLoop, kCFRunLoopDefaultMode);
15 
16  //一样
17  if (asyncSocket->writeStream)
18      CFWriteStreamScheduleWithRunLoop(asyncSocket->writeStream, runLoop, kCFRunLoopDefaultMode);
19 }

    这里可以看到，我们流的回调都是在这条流的常驻线程中，至于为什么要这么做，相信大家楼主看过AFNetworking系列文章的会明白。我们之后文章也会就这个框架线程的问题详细讨论的，这里就暂时不详细说明了。
    这里主要用了CFReadStreamScheduleWithRunLoop函数完成了runloop的注册:

  2. CFReadStreamScheduleWithRunLoop(asyncSocket->readStream, runLoop, kCFRunLoopDefaultMode);
CFWriteStreamScheduleWithRunLoop(asyncSocket->writeStream, runLoop, kCFRunLoopDefaultMode);

    这样，如果stream中有我们监听的事件发生了，就会在这个runloop中触发我们之前设置的读写回调函数。

我们完成了注册，接下来我们就需要打开stream了：

Stream相关方法四 -- 打开stream：

 1 //打开stream
 2 - (BOOL)openStreams
 3 {
 4     LogTrace();
 5 
 6     NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
 7     //断言读写stream都不会空
 8     NSAssert((readStream != NULL && writeStream != NULL), @"Read/Write stream is null");
 9 
10     //返回stream的状态
11 
12     CFStreamStatus readStatus = CFReadStreamGetStatus(readStream);
13     CFStreamStatus writeStatus = CFWriteStreamGetStatus(writeStream);
14 
15     //如果有任意一个没有开启
16     if ((readStatus == kCFStreamStatusNotOpen) || (writeStatus == kCFStreamStatusNotOpen))
17     {
18         LogVerbose(@"Opening read and write stream...");
19 
20         //开启
21         BOOL r1 = CFReadStreamOpen(readStream);
22         BOOL r2 = CFWriteStreamOpen(writeStream);
23 
24         //有一个开启失败
25         if (!r1 || !r2)
26         {
27             LogError(@"Error in CFStreamOpen");
28             return NO;
29         }
30     }
31 
32     return YES;
33 }

方法也很简单，通过CFReadStreamGetStatus函数，获取到当前stream的状态，判断没开启则调用CFReadStreamOpen函数去开启，如果开启失败，错误返回。

 到这里stream初始化相关的工作就做完了，接着我们还是回到本文方法十一 -- 连接成功后的初始化中：

其中第5条，我们谈到了设置socket的I/O模式为非阻塞，相信很多朋友对socket的I/O：同步、异步、阻塞、非阻塞。这四个概念有所混淆。
简单的来说，同步、异步是对于客户端而言的。比如我发起一个调用一个函数，我如果直接去调用，那么就是同步的，否则新开辟一个线程去做，那么对于当前线程而言就是异步的。
而阻塞和非阻塞是对于服务端而言。当服务端被客户端调用后，我如果立刻返回调用的结果（无论数据是否处理完）那么就是非阻塞的，又或者等待数据拿到并且处理完（总之一系列逻辑）再返回，那么这种情况就是阻塞的。

好了，有了这个概念，我们接下来看看Linux下的5种I/O模型:
1)阻塞I/O（blocking I/O）
2)非阻塞I/O （nonblocking I/O）
3) I/O复用(select 和poll) （I/O multiplexing）
4)信号驱动I/O （signal driven I/O (SIGIO)）
5)异步I/O （asynchronous I/O (the POSIX aio_functions)）

我们来简单谈谈这5种模型：
1）阻塞I/O：
简单举个例子，比如我们调用read()去读取消息，如果是在阻塞模式下，我们会一直等待，直到有消息到来为止。
很多小伙伴可能又要说了，这有什么不可以，我们新开辟一条线程，让它等着不就行了，看起来确实没什么不可以。
那是因为你仅仅是站在客户端的角度上来看。试想如果我们服务端也这么做，那岂不是有多少个socket连接，我们得开辟多少个线程去做阻塞IO？
2）非阻塞I/O
于是就有了非阻塞的概念，当我们去read()的时候，直接返回结果，这样在很大概率下，是并没有消息给我们读的。这时候函数就会错误返回-1，并将errno设置为 EWOULDBLOCK，意为IO并没有数据。
这时候就需要我们自己有一个机制，能知道什么时候有数据，在去调用read()。有一个很傻的方式就是不停的循环去调用这个函数，这样有数据来，我们第一时间就读到了。
3）I/O复用模式
I/O复用模式是阻塞I/O的改进版，它在read之前，会先去调用select去遍历所有的socket，看哪一个有消息。当然这个过程是阻塞的，直到有消息返回为止。然后在去调用read，阻塞的方式去读取从系统内核中去读取这条消息到进程中来。
4）信号驱动I/O
信号驱动I/O是一个半异步的I/O模式，它首先会调用一个系统sginal相关的函数，把socket和信号绑定起来，然后不管有没有消息直接返回(这一步非阻塞)。这时候系统内核会去检查socket是否有可用数据。有的话则发送该信号给进程，然后进程在去调用read阻塞式的从系统内核读取数据到进程中来（这一步阻塞）。
5）可能聪明的你已经想到了更好的解决方式，这就对了，这就是我们第5种IO模式：异步I/O ,它和第4步一样，也是调用sginal相关函数，把socket和信号绑定起来，同时绑定起来的还有一块数据缓冲区buffer。然后无论有没有数据直接返回（非阻塞）。而系统内核会去检查是否有可用数据，一旦有可用数据，则触发信号，并且把数据填充到我们之前提供的数据缓冲区buffer中。这样我们进程被信号触发，并且直接能从buffer中读取到数据，整个过程没有任何阻塞。
很显然，我们CocoaAyncSocket框架用的就是第5种I/O模式。

如果大家对I/O模式仍然感到疑惑，可以看看这篇文章：
socket阻塞与非阻塞，同步与异步、I/O模型

接着我们继续看本文方法十一 -- 连接成功后的初始化中第6条，读写source的初始化方法：

 本文方法十二 -- 初始化读写source：

  1 //初始化读写source
  2 - (void)setupReadAndWriteSourcesForNewlyConnectedSocket:(int)socketFD
  3 {
  4     //GCD source DISPATCH_SOURCE_TYPE_READ 会一直监视着 socketFD，直到有数据可读
  5     readSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, socketFD, 0, socketQueue);
  6     //_dispatch_source_type_write ：监视着 socketFD，直到写数据了
  7     writeSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, socketFD, 0, socketQueue);
  8 
  9     // Setup event handlers
 10 
 11     __weak GCDAsyncSocket *weakSelf = self;
 12 
 13 #pragma mark readSource的回调
 14 
 15     //GCD事件句柄  读，当socket中有数据流出现，就会触发这个句柄，全自动，不需要手动触发
 16     dispatch_source_set_event_handler(readSource, ^{ @autoreleasepool {
 17     #pragma clang diagnostic push
 18     #pragma clang diagnostic warning "-Wimplicit-retain-self"
 19 
 20         __strong GCDAsyncSocket *strongSelf = weakSelf;
 21         if (strongSelf == nil) return_from_block;
 22 
 23         LogVerbose(@"readEventBlock");
 24         //从readSource中，获取到数据长度，
 25         strongSelf->socketFDBytesAvailable = dispatch_source_get_data(strongSelf->readSource);
 26         LogVerbose(@"socketFDBytesAvailable: %lu", strongSelf->socketFDBytesAvailable);
 27 
 28         //如果长度大于0，开始读数据
 29         if (strongSelf->socketFDBytesAvailable > 0)
 30             [strongSelf doReadData];
 31         else
 32             //因为触发了，但是却没有可读数据，说明读到当前包边界了。做边界处理
 33             [strongSelf doReadEOF];
 34 
 35     #pragma clang diagnostic pop
 36     }});
 37 
 38     //写事件句柄
 39     dispatch_source_set_event_handler(writeSource, ^{ @autoreleasepool {
 40     #pragma clang diagnostic push
 41     #pragma clang diagnostic warning "-Wimplicit-retain-self"
 42 
 43         __strong GCDAsyncSocket *strongSelf = weakSelf;
 44         if (strongSelf == nil) return_from_block;
 45 
 46         LogVerbose(@"writeEventBlock");
 47         //标记为接受数据
 48         strongSelf->flags |= kSocketCanAcceptBytes;
 49         //开始写
 50         [strongSelf doWriteData];
 51 
 52     #pragma clang diagnostic pop
 53     }});
 54 
 55     // Setup cancel handlers
 56 
 57     __block int socketFDRefCount = 2;
 58 
 59     #if !OS_OBJECT_USE_OBJC
 60     dispatch_source_t theReadSource = readSource;
 61     dispatch_source_t theWriteSource = writeSource;
 62     #endif
 63 
 64     //读写取消的句柄
 65     dispatch_source_set_cancel_handler(readSource, ^{
 66     #pragma clang diagnostic push
 67     #pragma clang diagnostic warning "-Wimplicit-retain-self"
 68 
 69         LogVerbose(@"readCancelBlock");
 70 
 71         #if !OS_OBJECT_USE_OBJC
 72         LogVerbose(@"dispatch_release(readSource)");
 73         dispatch_release(theReadSource);
 74         #endif
 75 
 76         if (--socketFDRefCount == 0)
 77         {
 78             LogVerbose(@"close(socketFD)");
 79             //关闭socket
 80             close(socketFD);
 81         }
 82 
 83     #pragma clang diagnostic pop
 84     });
 85 
 86     dispatch_source_set_cancel_handler(writeSource, ^{
 87     #pragma clang diagnostic push
 88     #pragma clang diagnostic warning "-Wimplicit-retain-self"
 89 
 90         LogVerbose(@"writeCancelBlock");
 91 
 92         #if !OS_OBJECT_USE_OBJC
 93         LogVerbose(@"dispatch_release(writeSource)");
 94         dispatch_release(theWriteSource);
 95         #endif
 96 
 97         if (--socketFDRefCount == 0)
 98         {
 99             LogVerbose(@"close(socketFD)");
100             //关闭socket
101             close(socketFD);
102         }
103 
104     #pragma clang diagnostic pop
105     });
106 
107     // We will not be able to read until data arrives.
108     // But we should be able to write immediately.
109 
110     //设置未读数量为0
111     socketFDBytesAvailable = 0;
112     //把读挂起的状态移除
113     flags &= ~kReadSourceSuspended;
114 
115     LogVerbose(@"dispatch_resume(readSource)");
116     //开启读source
117     dispatch_resume(readSource);
118 
119     //标记为当前可接受数据
120     flags |= kSocketCanAcceptBytes;
121     //先把写source标记为挂起
122     flags |= kWriteSourceSuspended;
123 }

这个方法初始化了读写source，这个方法主要是GCD source运用，如果有对这部分知识有所疑问，可以看看宜龙大神这篇:GCD高级用法。
这里GCD Source相关的主要是下面这3个函数： 

//创建source
dispatch_source_create(dispatch_source_type_t type,
    uintptr_t handle,
    unsigned long mask,
    dispatch_queue_t _Nullable queue);
//为source设置事件句柄
dispatch_source_set_event_handler(dispatch_source_t source,
    dispatch_block_t _Nullable handler);
//为source设置取消句柄
dispatch_source_set_cancel_handler(dispatch_source_t source,
    dispatch_block_t _Nullable handler);

相信大家用至少用过GCD定时器，接触过这3个函数，这里创建source的函数，根据参数type的不同，可以处理不同的事件：

这里我们用的是DISPATCH_SOURCE_TYPE_READ和DISPATCH_SOURCE_TYPE_WRITE这两个类型。标识如果handle如果有可读或者可写数据时，会触发我们的事件句柄。

  • 而这里初始化的读写事件句柄内容也很简单，就是去读写数据。
  • 而取消句柄也就是去关闭socket。
  • 初始化完成后，我们开启了readSource，一旦有数据过来就触发了我们readSource事件句柄，就可以去监听的socket所分配的缓冲区中去读取数据了，而wirteSource初始化完是挂起的。
  • 除此之外我们还初始化了当前source的状态，用于我们后续的操作。

至此我们客户端的整个Connect流程结束了，用一张图来概括总结一下吧：

整个客户端连接的流程大致如上图，当然远不及于此，这里我们对地址做了IPV4和IPV6的兼容处理，对一些使用socket而产生的网络错误导致进程退出的容错处理。以及在这个过程中，socketQueue、代理queue、全局并发queue和stream常驻线程的管理调度等等。

当然其中绝大部分操作都是在socketQueue中进行的。而在socketQueue中，我们也分为两种操作dispatch_sync和dispatch_async。
因为socketQueue本身就是一个串行queue，所以我们所有的操作都在这个queue中进行保证了线程安全，而需要阻塞后续行为的操作，我们用了sync的方式。其实这样使用sync是及其容易死锁的，但是作者每次在调用sync之前都调用了这么一行判断：

if (dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey))

判断当前队列是否就是这个socketQueue队列，如果是则直接调用，否则就用sync的方式提交到这个queue中去执行。这种防死锁的方式，你学到了么？

接着我们来讲讲服务端Accept流程：

整个流程还是相对Connect来说还是十分简单的，因为这个方法很长，而且大多数是我们直接连接讲到过得内容，所以我省略了一部分的代码，只把重要的展示出来，大家可以参照着源码看。

  1 //监听端口起点
  2 - (BOOL)acceptOnPort:(uint16_t)port error:(NSError **)errPtr
  3 {
  4      return [self acceptOnInterface:nil port:port error:errPtr];
  5 }
  6 
  7 - (BOOL)acceptOnInterface:(NSString *)inInterface port:(uint16_t)port error:(NSError **)errPtr
  8 {
  9      LogTrace();
 10 
 11      // Just in-case interface parameter is immutable.
 12     //防止参数被修改
 13      NSString *interface = [inInterface copy];
 14 
 15      __block BOOL result = NO;
 16      __block NSError *err = nil;
 17 
 18      // CreateSocket Block
 19      // This block will be invoked within the dispatch block below.
 20      //创建socket的Block
 21      int(^createSocket)(int, NSData*) = ^int (int domain, NSData *interfaceAddr) {
 22 
 23         //创建TCP的socket
 24           int socketFD = socket(domain, SOCK_STREAM, 0);
 25 
 26           //一系列错误判断
 27           ...
 28           // Bind socket
 29         //用本地地址去绑定
 30           status = bind(socketFD, (const struct sockaddr *)[interfaceAddr bytes], (socklen_t)[interfaceAddr length]);
 31 
 32           //监听这个socket
 33         //第二个参数是这个端口下维护的socket请求队列，最多容纳的用户请求数。
 34           status = listen(socketFD, 1024);
 35           return socketFD;
 36      };
 37 
 38      // Create dispatch block and run on socketQueue
 39 
 40      dispatch_block_t block = ^{ @autoreleasepool {
 41 
 42          //一系列错误判断
 43           ...
 44 
 45         //判断ipv4 ipv6是否支持
 46           ...
 47 
 48         //得到本机的IPV4 IPV6的地址
 49           [self getInterfaceAddress4:&interface4 address6:&interface6 fromDescription:interface port:port];
 50           ...
 51 
 52         //判断可以用IPV4还是6进行请求
 53           ...
 54 
 55           // Create accept sources
 56           //创建接受连接被触发的source
 57           if (enableIPv4)
 58           {
 59             //接受连接的source
 60                accept4Source = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, socket4FD, 0, socketQueue);
 61 
 62 
 63 
 64             //事件句柄
 65                dispatch_source_set_event_handler(accept4Source, ^{ @autoreleasepool {
 66 
 67                 //拿到数据，连接数
 68                     unsigned long numPendingConnections = dispatch_source_get_data(acceptSource);
 69 
 70                     LogVerbose(@"numPendingConnections: %lu", numPendingConnections);
 71 
 72                 //循环去接受这些socket的事件(一次触发可能有多个连接)
 73                     while ([strongSelf doAccept:socketFD] && (++i < numPendingConnections));
 74 
 75                }});
 76 
 77                //取消句柄
 78                dispatch_source_set_cancel_handler(accept4Source, ^{
 79                 //...
 80                 //关闭socket
 81                     close(socketFD);
 82 
 83                });
 84 
 85             //开启source
 86                dispatch_resume(accept4Source);
 87           }
 88 
 89         //ipv6一样
 90           ...
 91 
 92     //在scoketQueue中同步做这些初始化。
 93      if (dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey))
 94           block();
 95      else
 96           dispatch_sync(socketQueue, block);
 97 
 98      //...错误判断
 99      //返回结果
100      return result;
101 }

这个方法省略完仍然有这么长，它主要做了这两件事（篇幅原因，尽量精简）：

  1. 创建本机地址、创建socket、绑定端口、监听端口。
  2. 创建了一个GCD Source，来监听这个socket读source，这样连接事件一发生，就会触发我们的事件句柄。接着我们调用了doAccept:方法循环去接受所有的连接。

接着我们来看这个接受连接的方法（同样省略了一部分不那么重要的代码）：

 1 //连接接受的方法
 2 - (BOOL)doAccept:(int)parentSocketFD
 3 {
 4      LogTrace();
 5 
 6      int socketType;
 7      int childSocketFD;
 8      NSData *childSocketAddress;
 9 
10     //IPV4
11      if (parentSocketFD == socket4FD)
12      {
13           socketType = 0;
14 
15           struct sockaddr_in addr;
16           socklen_t addrLen = sizeof(addr);
17           //调用接受，得到接受的子socket
18           childSocketFD = accept(parentSocketFD, (struct sockaddr *)&addr, &addrLen);
19           //NO说明没有连接
20           if (childSocketFD == -1)
21           {
22                LogWarn(@"Accept failed with error: %@", [self errnoError]);
23                return NO;
24           }
25           //子socket的地址数据
26           childSocketAddress = [NSData dataWithBytes:&addr length:addrLen];
27      }
28     //一样
29      else if (parentSocketFD == socket6FD)
30      {
31           ...
32      }
33     //unix domin socket 一样
34      else // if (parentSocketFD == socketUN)
35      {
36           ...
37      }
38 
39      //socket 配置项的设置... 和connect一样
40 
41     //响应代理
42      if (delegateQueue)
43      {
44           __strong id theDelegate = delegate;
45           //代理队列中调用
46           dispatch_async(delegateQueue, ^{ @autoreleasepool {
47 
48                // Query delegate for custom socket queue
49 
50                dispatch_queue_t childSocketQueue = NULL;
51 
52             //判断是否实现了为socket 生成一个新的SocketQueue，是的话拿到新queue
53                if ([theDelegate respondsToSelector:@selector(newSocketQueueForConnectionFromAddress:onSocket:)])
54                {
55                     childSocketQueue = [theDelegate newSocketQueueForConnectionFromAddress:childSocketAddress
56                                                                                   onSocket:self];
57                }
58 
59                // Create GCDAsyncSocket instance for accepted socket
60                //新创建一个本类实例，给接受的socket
61                GCDAsyncSocket *acceptedSocket = [[[self class] alloc] initWithDelegate:theDelegate
62                                                                                        delegateQueue:delegateQueue
63                                                                                           socketQueue:childSocketQueue];
64                //IPV4 6 un
65                if (socketType == 0)
66                     acceptedSocket->socket4FD = childSocketFD;
67                else if (socketType == 1)
68                     acceptedSocket->socket6FD = childSocketFD;
69                else
70                     acceptedSocket->socketUN = childSocketFD;
71                //标记开始 并且已经连接
72                acceptedSocket->flags = (kSocketStarted | kConnected);
73 
74                // Setup read and write sources for accepted socket
75                //初始化读写source
76                dispatch_async(acceptedSocket->socketQueue, ^{ @autoreleasepool {
77 
78                     [acceptedSocket setupReadAndWriteSourcesForNewlyConnectedSocket:childSocketFD];
79                }});
80 
81             //判断代理是否实现了didAcceptNewSocket方法，把我们新创建的socket返回出去
82                if ([theDelegate respondsToSelector:@selector(socket:didAcceptNewSocket:)])
83                {
84                     [theDelegate socket:self didAcceptNewSocket:acceptedSocket];
85                }
86 
87           }});
88      }
89      return YES;
90 }
  • 这个方法很简单，核心就是调用下面这个函数，去接受连接，并且拿到一个新的socket
  • childSocketFD = accept(parentSocketFD, (struct sockaddr *)&addr, &addrLen);
  • 然后调用了newSocketQueueForConnectionFromAddress:onSocket:这个代理，可以为新的socket重新设置一个socketQueue。
  • 接着我们用这个Socket重新创建了一个GCDAsyncSocket实例，然后调用我们的代理didAcceptNewSocket方法，把这个实例给传出去了。
  • 这里需要注意的是，我们调用didAcceptNewSocket代理方法传出去的实例我们需要自己保留，不然就会被释放掉，那么这个与客户端的连接也就断开了。
  • 同时我们还初始化了这个新socket的读写source，这一步完全和connect中一样，调用同一个方法，这样如果有读写数据，就会触发这个新的socket的source了。

建立连接之后的无数个新的socket，都是独立的，它们处理读写连接断开的逻辑就和客户端socket完全一样了。
而我们监听本机端口的那个socket始终只有一个，这个用来监听触发socket连接，并返回创建我们这无数个新的socket实例。

作为服务端的Accept流程就这么结束了，因为篇幅原因，所以尽量精简了一些细节的处理，不过这些处理在Connect中也是反复出现的，所以基本无伤大雅。如果大家会感到困惑，建议下载github中的源码注释，对照着再看一遍，相信会有帮助的。

接着我们来讲讲Unix Domin Socket建立本地进程通信流程：

基本上这个流程，比上述任何流程还要简单，简单的到即使不简化代码，也没多少行（当然这是建立在客户端Connect流程已经实现了很多公用方法的基础上）。

接着进入正题，我们来看看它发起连接的方法：

 1 //连接本机的url上，IPC，进程间通信
 2 - (BOOL)connectToUrl:(NSURL *)url withTimeout:(NSTimeInterval)timeout error:(NSError **)errPtr;
 3 {
 4     LogTrace();
 5 
 6     __block BOOL result = NO;
 7     __block NSError *err = nil;
 8 
 9     dispatch_block_t block = ^{ @autoreleasepool {
10 
11         //判断长度
12         if ([url.path length] == 0)
13         {
14             NSString *msg = @"Invalid unix domain socket url.";
15             err = [self badParamError:msg];
16 
17             return_from_block;
18         }
19 
20         // Run through standard pre-connect checks
21         //前置的检查
22         if (![self preConnectWithUrl:url error:&err])
23         {
24             return_from_block;
25         }
26 
27         // We've made it past all the checks.
28         // It's time to start the connection process.
29 
30         flags |= kSocketStarted;
31 
32         // Start the normal connection process
33 
34         NSError *connectError = nil;
35         //调用另一个方法去连接
36         if (![self connectWithAddressUN:connectInterfaceUN error:&connectError])
37         {
38             [self closeWithError:connectError];
39 
40             return_from_block;
41         }
42 
43         [self startConnectTimeout:timeout];
44 
45         result = YES;
46     }};
47 
48     //在socketQueue中同步执行
49     if (dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey))
50         block();
51     else
52         dispatch_sync(socketQueue, block);
53 
54     if (result == NO)
55     {
56         if (errPtr)
57             *errPtr = err;
58     }
59 
60     return result;
61 }

连接方法非常简单，就只是做了一些错误的处理，然后调用了其他的方法，包括一个前置检查，这检查中会去判断各种参数是否正常，如果正常会返回YES，并且把url转换成Uinix domin socket地址的结构体，赋值给我们的属性connectInterfaceUN。
接着调用了connectWithAddressUN方法去发起连接。

我们接着来看看这个方法：

 1 //连接Unix域服务器
 2 - (BOOL)connectWithAddressUN:(NSData *)address error:(NSError **)errPtr
 3 {
 4      LogTrace();
 5 
 6      NSAssert(dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey), @"Must be dispatched on socketQueue");
 7 
 8      // Create the socket
 9 
10      int socketFD;
11 
12      LogVerbose(@"Creating unix domain socket");
13 
14     //创建本机socket
15      socketUN = socket(AF_UNIX, SOCK_STREAM, 0);
16 
17      socketFD = socketUN;
18 
19      if (socketFD == SOCKET_NULL)
20      {
21           if (errPtr)
22                *errPtr = [self errnoErrorWithReason:@"Error in socket() function"];
23 
24           return NO;
25      }
26 
27      // Bind the socket to the desired interface (if needed)
28 
29      LogVerbose(@"Binding socket...");
30 
31      int reuseOn = 1;
32     //设置可复用
33      setsockopt(socketFD, SOL_SOCKET, SO_REUSEADDR, &reuseOn, sizeof(reuseOn));
34 
35      // Prevent SIGPIPE signals
36 
37      int nosigpipe = 1;
38     //进程终止错误信号禁止
39      setsockopt(socketFD, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe, sizeof(nosigpipe));
40 
41      // Start the connection process in a background queue
42 
43      int aStateIndex = stateIndex;
44 
45      dispatch_queue_t globalConcurrentQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
46      dispatch_async(globalConcurrentQueue, ^{
47 
48           const struct sockaddr *addr = (const struct sockaddr *)[address bytes];
49         //并行队列调用连接
50           int result = connect(socketFD, addr, addr->sa_len);
51           if (result == 0)
52           {
53                dispatch_async(socketQueue, ^{ @autoreleasepool {
54                     //连接成功的一些状态初始化
55                     [self didConnect:aStateIndex];
56                }});
57           }
58           else
59           {
60                // 失败的处理
61                perror("connect");
62                NSError *error = [self errnoErrorWithReason:@"Error in connect() function"];
63 
64                dispatch_async(socketQueue, ^{ @autoreleasepool {
65 
66                     [self didNotConnect:aStateIndex error:error];
67                }});
68           }
69      });
70 
71      LogVerbose(@"Connecting...");
72 
73      return YES;
74 }

主要部分基本和客户端连接相同，并且简化了很多，调用了这一行完成了连接：

int result = connect(socketFD, addr, addr->sa_len);

同样也和客户端一样，在连接成功之后去调用下面这个方法完成了一些资源的初始化：

[self didConnect:aStateIndex];

基本上连接就这么两个方法了（当然我们省略了一些细节），看完客户端的连接之后，到这就变得非常简单了。

接着我们来看看uinix domin socket作为服务端Accept。

这个Accpet，基本和我们普通Socket服务端的Accept相同。

 1 //接受一个Url，uniex domin socket 做为服务端
 2 - (BOOL)acceptOnUrl:(NSURL *)url error:(NSError **)errPtr;
 3 {
 4      LogTrace();
 5 
 6      __block BOOL result = NO;
 7      __block NSError *err = nil;
 8 
 9     //基本和正常的socket accept一模一样
10      // CreateSocket Block
11      // This block will be invoked within the dispatch block below.
12      //生成一个创建socket的block，创建、绑定、监听
13      int(^createSocket)(int, NSData*) = ^int (int domain, NSData *interfaceAddr) {
14 
15           //creat  socket 
16           ...
17           // Set socket options
18 
19           ...
20           // Bind socket
21 
22           ...
23 
24           // Listen
25           ...
26      };
27 
28      // Create dispatch block and run on socketQueue
29      //错误判断
30      dispatch_block_t block = ^{ @autoreleasepool {
31 
32           //错误判断
33           ...
34 
35 
36           //判断是否有这个url路径是否正确
37           ...
38 
39           //调用上面的Block创建socket，并且绑定监听。
40           ...
41 
42           //创建接受连接的source
43           acceptUNSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, socketUN, 0, socketQueue);
44 
45           int socketFD = socketUN;
46           dispatch_source_t acceptSource = acceptUNSource;
47           //事件句柄，和accpept一样
48           dispatch_source_set_event_handler(acceptUNSource, ^{ @autoreleasepool {
49                //循环去接受所有的每一个连接
50                ...
51           }});
52 
53           //取消句柄
54           dispatch_source_set_cancel_handler(acceptUNSource, ^{
55 
56                //关闭socket
57                close(socketFD);
58           });
59 
60           LogVerbose(@"dispatch_resume(accept4Source)");
61           dispatch_resume(acceptUNSource);
62 
63           flags |= kSocketStarted;
64 
65           result = YES;
66      }};
67 
68      if (dispatch_get_specific(IsOnSocketQueueOrTargetQueueKey))
69           block();
70      else
71           dispatch_sync(socketQueue, block);
72      //填充错误
73      if (result == NO)
74      {
75           LogInfo(@"Error in accept: %@", err);
76 
77           if (errPtr)
78                *errPtr = err;
79      }
80 
81      return result; 
82 }

因为代码基本雷同，所以我们省略了大部分代码，大家可以参照着之前的讲解或者源码去理解。这里和普通服务端socket唯一的区别就是，这里服务端绑定的地址是unix domin socket类型的地址，它是一个结构体，里面包含的是我们进行进程通信的纽带-一个本机文件路径。
所以这里服务端简单来说就是绑定的这个文件路径，当这个文件路径有数据可读（即有客户端连接到达）的时候，会触发初始化的source事件句柄，我们会去循环的接受所有的连接，并且新生成一个socket实例，这里和普通的socket完全一样。

就这样我们所有的连接方式已经讲完了，后面这两种方式，为了节省篇幅，确实讲的比较粗略，但是核心的部分都有提到。
另外如果你有理解客户端的Connect流程，那么理解起来应该没有什么问题，这两个流程比前者可简化太多了。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/francisblogs/p/6829722.html