NSOperationQueue GUNStep源码

NSOperation GUNStep源码
NSInvocationOperation GUNStep源码
NSOperationQueue GUNStep源码

首先是 initialize 方法,在类第一次使用的时候,创建一个实例当做主队列.

+ (void) initialize
{
  if (nil == mainQueue)
    {
      mainQueue = [self new];
    }
}

获取当前队列

+ (id) currentQueue
{
  if ([NSThread isMainThread]) // 如果是主线程
    {
      return mainQueue; // 返回主队列
    }
// 返回当前线程的 队列信息
  return [[[NSThread currentThread] threadDictionary] objectForKey: threadKey]; 
}

init方法

- (id) init
{
  if ((self = [super init]) != nil)
    {
      GS_CREATE_INTERNAL(NSOperationQueue);
      internal->suspended = NO; 默认没有暂停
      internal->count = NSOperationQueueDefaultMaxConcurrentOperationCount; // 最大并发数
      internal->operations = [NSMutableArray new]; // 操作s
      internal->starting = [NSMutableArray new]; // 正在进行的操作s
      internal->waiting = [NSMutableArray new]; // 等待的操作s
      internal->lock = [NSRecursiveLock new];
      [internal->lock setName:
        [NSString stringWithFormat: @"lock-for-op-%p", self]];
      internal->cond = [[NSConditionLock alloc] initWithCondition: 0];
      [internal->cond setName:
        [NSString stringWithFormat: @"cond-for-op-%p", self]];
    }
  return self;
}

添加operation

- (void) addOperation: (NSOperation *)op
{// 异常处理
  if (op == nil || NO == [op isKindOfClass: [NSOperation class]])
    {
      [NSException raise: NSInvalidArgumentException
          format: @"[%@-%@] object is not an NSOperation",
    NSStringFromClass([self class]), NSStringFromSelector(_cmd)];
    }
  [internal->lock lock];
// 如果任务还没添加到数组并且任务还没开始执行
  if (NSNotFound == [internal->operations indexOfObjectIdenticalTo: op]&& NO == [op isFinished]) 
    {
      [op addObserver: self
       forKeyPath: @"isReady"
          options: NSKeyValueObservingOptionNew
          context: isReadyCtxt]; // 监听是否准备好
      [self willChangeValueForKey: @"operations"];
      [self willChangeValueForKey: @"operationCount"];
      [internal->operations addObject: op];
      [self didChangeValueForKey: @"operationCount"];
      [self didChangeValueForKey: @"operations"];
      if (YES == [op isReady]) // 如果准备好了,手动发送KVO通知
    {
      [self observeValueForKeyPath: @"isReady"
                  ofObject: op
                change: nil
                   context: isReadyCtxt];
    }
    }
  [internal->lock unlock];
}

添加操作,并是否阻塞当前线程直到结束

- (void) addOperations: (NSArray *)ops
     waitUntilFinished: (BOOL)shouldWait
{
  NSUInteger    total;
  NSUInteger    index;

    // 异常判断
  if (ops == nil || NO == [ops isKindOfClass: [NSArray class]])
    {
      [NSException raise: NSInvalidArgumentException
          format: @"[%@-%@] object is not an NSArray",
    NSStringFromClass([self class]), NSStringFromSelector(_cmd)];
    }
    // 总的任务数量
  total = [ops count];
    
  if (total > 0)
    {
      BOOL      invalidArg = NO;
      NSUInteger    toAdd = total;
        // 创建一个新的数组存储operation
      GS_BEGINITEMBUF(buf, total, id)

        // 获取传进来的 ops 放到 buf里面,之后操作 buf就可以了
      [ops getObjects: buf];
        //
      for (index = 0; index < total; index++)
    {
      NSOperation   *op = buf[index];

        // 如果有任务 不是 NSOperation,就默认没有需要添加的任务,后续抛出异常
      if (NO == [op isKindOfClass: [NSOperation class]])
        {
          invalidArg = YES;
          toAdd = 0;
          break;
        }
        // 去除已经完成的任务
      if (YES == [op isFinished])
        {
          buf[index] = nil;
          toAdd--;
        }
    }
        // 如果要添加的大于0
      if (toAdd > 0)
    {
        // 加锁
          [internal->lock lock];
      [self willChangeValueForKey: @"operationCount"];
      [self willChangeValueForKey: @"operations"];
        
      for (index = 0; index < total; index++)
        {
          NSOperation   *op = buf[index];

          if (op == nil)
        {
          continue;     // Not added
        }
            // 如果已经在 operations 数组里了,就 任务数-1 continue
          if (NSNotFound
        != [internal->operations indexOfObjectIdenticalTo: op])
        {
          buf[index] = nil; // Not added
          toAdd--;
          continue;
        }
            
            //监听 op 的 isReady 状态
          [op addObserver: self
           forKeyPath: @"isReady"
              options: NSKeyValueObservingOptionNew
              context: isReadyCtxt];
            // 添加操作
          [internal->operations addObject: op];
          if (NO == [op isReady])// 如果不是 ready 就删除,后续所有ready状态的都会直接调用kvo监听
        {
          buf[index] = nil; // Not yet ready
        }
        }
      [self didChangeValueForKey: @"operationCount"];
      [self didChangeValueForKey: @"operations"];
        
      for (index = 0; index < total; index++)
        {
          NSOperation   *op = buf[index];

          if (op != nil)
        {
            // 直接调用kvo监听
          [self observeValueForKeyPath: @"isReady"
                      ofObject: op
                    change: nil
                       context: isReadyCtxt];
        }
        }
          [internal->lock unlock];
    }
      GS_ENDITEMBUF()
      if (YES == invalidArg)// 抛出异常
    {
      [NSException raise: NSInvalidArgumentException
        format: @"[%@-%@] object at index %"PRIuPTR" is not an NSOperation",
        NSStringFromClass([self class]), NSStringFromSelector(_cmd),
        index];
    }
    }
    // 如果需要监听到完成,就去给op加条件锁
  if (YES == shouldWait)
    {
      [self waitUntilAllOperationsAreFinished];
    }
}

给op加条件锁

- (void) waitUntilAllOperationsAreFinished
{
  NSOperation   *op;

  [internal->lock lock];
  while ((op = [internal->operations lastObject]) != nil)
    { //从后往前对所有的op加条件锁, 
      [op retain];
      [internal->lock unlock];
      [op waitUntilFinished];
      [op release];
      [internal->lock lock];
    }
  [internal->lock unlock];
}

设置最大线程数

- (void) setMaxConcurrentOperationCount: (NSInteger)cnt
{
// 异常判断
  if (cnt < 0
    && cnt != NSOperationQueueDefaultMaxConcurrentOperationCount)
    {
      [NSException raise: NSInvalidArgumentException
          format: @"[%@-%@] cannot set negative (%"PRIdPTR") count",
    NSStringFromClass([self class]), NSStringFromSelector(_cmd), cnt];
    }
  [internal->lock lock];
//修改
  if (cnt != internal->count)
    {
      [self willChangeValueForKey: @"maxConcurrentOperationCount"];
      internal->count = cnt;
      [self didChangeValueForKey: @"maxConcurrentOperationCount"];
    }
  [internal->lock unlock];
// 执行
  [self _execute];
}

KVO监听回调,在添加Oeration的时候

- (void) observeValueForKeyPath: (NSString *)keyPath
               ofObject: (id)object
                         change: (NSDictionary *)change
                        context: (void *)context
{
  /* We observe three properties in sequence ...
   * isReady (while we wait for an operation to be ready)
   * queuePriority (when priority of a ready operation may change)
   * isFinished (to see if an executing operation is over).
   */
  if (context == isFinishedCtxt)
    {
      [internal->lock lock];
      internal->executing--;
      [object removeObserver: self forKeyPath: @"isFinished"];
      [internal->lock unlock];
      [self willChangeValueForKey: @"operations"];
      [self willChangeValueForKey: @"operationCount"];
      [internal->lock lock];
      [internal->operations removeObjectIdenticalTo: object];
      [internal->lock unlock];
      [self didChangeValueForKey: @"operationCount"];
      [self didChangeValueForKey: @"operations"];
    }
  else if (context == queuePriorityCtxt || context == isReadyCtxt) // 如果是准备好了
    {
      NSInteger pos;

      [internal->lock lock];
      if (context == queuePriorityCtxt)
        {
          [internal->waiting removeObjectIdenticalTo: object];
        }
      if (context == isReadyCtxt)
        {
          [object removeObserver: self forKeyPath: @"isReady"]; // 移除监听准备状态
          [object addObserver: self
                   forKeyPath: @"queuePriority"
                      options: NSKeyValueObservingOptionNew
                      context: queuePriorityCtxt]; // 监听优先级
        }
      pos = [internal->waiting insertionPosition: object
                                   usingFunction: sortFunc
                                         context: 0]; // 获得插入的顺序
      [internal->waiting insertObject: object atIndex: pos]; // 插入到waiting里
        
      [internal->lock unlock];
    }
  [self _execute]; // 开始执行?
}

执行任务代码,判断等待的任务和当前执行数,如果符合条件,取出来任务,同步或者开辟线程执行

/* Check for operations which can be executed and start them.
 */
- (void) _execute
{
  NSInteger max;

  [internal->lock lock];

  max = [self maxConcurrentOperationCount]; // 最大执行数
  if (NSOperationQueueDefaultMaxConcurrentOperationCount == max)
    {
      max = maxConcurrent; // 线程数 200
    }

  while (NO == [self isSuspended]
    && max > internal->executing
    && [internal->waiting count] > 0)
    { // 如果没有挂起,并且有等待的任务,并且正在执行的线程数小于 最大线程数
      NSOperation   *op;

      /* Take the first operation from the queue and start it executing.
       * We set ourselves up as an observer for the operating finishing
       * and we keep track of the count of operations we have started,
       * but the actual startup is left to the NSOperation -start method.
       */
      op = [internal->waiting objectAtIndex: 0]; // 取代正在等待的任务
      [internal->waiting removeObjectAtIndex: 0];
      [op removeObserver: self forKeyPath: @"queuePriority"]; // 移除 queuePriority 监听
      [op addObserver: self
       forKeyPath: @"isFinished"
          options: NSKeyValueObservingOptionNew
          context: isFinishedCtxt]; // 监听完成
      internal->executing++; // 正在执行数 +1
      if (YES == [op isConcurrent]) // 如果是并发,直接执行
    {
          [op start]; // 执行任务
    }
      else
    {
      NSUInteger    pending;

      [internal->cond lock];
      pending = [internal->starting count];
      [internal->starting addObject: op]; // 添加到正在执行

      /* Create a new thread if all existing threads are busy and
       * we haven't reached the pool limit.
       */
      if (0 == internal->threadCount
        || (pending > 0 && internal->threadCount < POOL))
        {
          internal->threadCount++; // 线程数 + 1
          [NSThread detachNewThreadSelector: @selector(_thread)
                       toTarget: self
                     withObject: nil]; // 创建线程并且执行
        }
      /* Tell the thread pool that there is an operation to start.
       */
      [internal->cond unlockWithCondition: 1];
    }
    }
  [internal->lock unlock];
}

在子线程的执行任务的代码: 获取标记为start的任务,开始执行,直到start数组里没有任务了,退出线程
// 在子线程的执行任务的代码

- (void) _thread
{
  NSAutoreleasePool *pool = [NSAutoreleasePool new]; // 自动释放池

  [[[NSThread currentThread] threadDictionary] setObject: self
                                                  forKey: threadKey]; // 设置线程的 operationQueue 用以 currentQueue 去获取到
  for (;;)
    {
      NSOperation   *op;
      NSDate        *when;
      BOOL      found;
        
      when = [[NSDate alloc] initWithTimeIntervalSinceNow: 5.0];// 五秒?
      found = [internal->cond lockWhenCondition: 1 beforeDate: when]; // 条件锁等待五秒?
      RELEASE(when);
      if (NO == found)
    {
      break;    // Idle for 5 seconds ... exit thread. // 如果没有等待到任务,就退出
    }

      if ([internal->starting count] > 0) // 如果有正在执行的任务
    {
          op = RETAIN([internal->starting objectAtIndex: 0]);// 获取第一个正在执行的任务
      [internal->starting removeObjectAtIndex: 0]; // 移除任务
    }
      else
    {
      op = nil; // 没有正在执行的任务 op 置为nil
    }

      if ([internal->starting count] > 0) // 如果还有正在执行的任务
    {
      // Signal any other idle threads,
          [internal->cond unlockWithCondition: 1]; // 解锁 1 条件 , 可以再次获取,再次执行,这个主要是解锁给其他线程用的,其他线程还可以取到任务
    }
      else
    {
      // There are no more operations starting.
          [internal->cond unlockWithCondition: 0]; // 解锁 0 条件 , 这样其他线程可以等5秒,如果5秒没有任务,直接结束线程
    }

      if (nil != op) // 如果取到了任务,就开始执行任务
    {
          NS_DURING
        {
          NSAutoreleasePool *opPool = [NSAutoreleasePool new];

              [NSThread setThreadPriority: [op threadPriority]]; // 设置线程优先级
              [op start]; // 开始执行 op
          RELEASE(opPool);
        }
          NS_HANDLER
        {
          NSLog(@"Problem running operation %@ ... %@",
        op, localException);
        }
          NS_ENDHANDLER
      [op _finish]; // 执行完成
          RELEASE(op);
    }
    }

  [[[NSThread currentThread] threadDictionary] removeObjectForKey: threadKey]; // 移除线程的队列属性
  [internal->lock lock];
  internal->threadCount--; // 线程数量-1
  [internal->lock unlock];
  RELEASE(pool);
  [NSThread exit]; // 退出线程
}

总结; NSOperationQueue 负责监听Operation 的 ready,queuePriority,finlish.状态.
内部维护了三个数组operations,waiting,starting.
负责创建线程,执行操作.

但GUNStep的代码并没有针对 mainQueue 做特殊的处理.