基于pthread创建ThreadPool(线程池)和QispatchQueue

ThreadPool简介

我们都知道多线程有很多好处，同时创建管理维护等也有很多成本，大致有以下几个缺点：

  1. 空间成本：每个线程占用的空间512kb，意味着更多的线程需要更多的内存空间
  2. 时间成本：创建一个线程大约90毫秒
  3. 线程间的通信，多线程的数据共享，有些数据是在多个线程间共享的,需要防止线程死锁情况的发生
  4. 如果有大量的线程,会影响性能,因为操作系统需要在它们之间切换

线程池可以节省不断创建线程和释放线程的时间，所谓线程池就是预先创建好一组线程，线程池有一个任务队列，当有一个新任务需要执行时将该任务放入到任务队列中，线程池中空闲的线程会从该队列中取出任务并执行，这样做的好处就是减少了创建线程和销毁线程带来的效率问题，在任务处理时间比较短的时候这个好处非常显著，可以提升任务处理的效率。
Pool的任务队列是一个优先级任务队列，这意味着新加入的任务如果优先级比较高的话则将可能会被线程池中空闲线程优先考虑执行。

线程池在多线程编程中经常要用到，其基本模型仍是生产者/消费者模型，线程池一般由线程池管理器（ThreadPool），工作线程（PoolWorker），任务（ Task），任务队列（TaskQueue）四部分组成，其中

ThreadPool：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；
PoolWorker：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；
Task：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
taskQueue：用于存放没有处理的任务。提供一种缓冲机制。

简单的工作流程图如下：

Screen Shot 2018-07-28 at 3.28.16 PM.png

pthread简介

POSIX线程（英语：POSIX Threads，常被缩写为Pthreads）是POSIX的线程标准，定义了创建和操纵线程的一套API。
pthread是一套通用的多线程的API，可以在Unix / Linux / Windows 等系统跨平台使用，使用C语言编写，需要程序员自己管理线程的生命周期，使用难度较大。

可能很多iOS没有听说过pthread，确实苹果为我们提供了NSThread，使用起来比pthread更加面向对象，简单易用，可以直接操作线程对象。不过也需要需要程序员自己管理线程的生命周期(主要是创建)，我们在开发的过程中偶尔使用NSThread。甚至封装了更好用的GCD和NSOperation，我们在平常开发中是没必要去使用pthread的，

结构体定义

Condition（任务锁）

线程池可能会被外部或内部的多个线程同时访问，因此必须保证其线程安全。除了保证读写操作互斥外，该锁还要支持 wait和signal操作，所以这里不要使用NSLock，NSCondition是可以满足我们的要求的，但是这一次我们希望尽可能都用pthread的接口。所以这次我们选择pthread_mutex_t和pthread_cond_t，我们把他们封装成Conditon让它具备类似NSCondition的功能，类似这样：

class Condition{
    
    private var mutex = pthread_mutex_t()
    private var cont = pthread_cond_t()
    
    init() {
        pthread_mutex_init(&self.mutex, nil)
        pthread_cond_init(&self.cont, nil)
    }
    
    deinit {
        pthread_mutex_destroy(&mutex)
        pthread_cond_destroy(&cont)
    }
    
    func lock() {
        pthread_mutex_lock(&mutex)
    }
    
    func unlock() {
        pthread_mutex_unlock(&mutex)
    }
    
    func wait() {
        pthread_cond_wait(&cont, &mutex)
    }
    
    func signal() {
        pthread_cond_broadcast(&cont)
    }
}

Lock（互斥锁）

我们可以用NSLok，同Condition一样我们选择自己封装

class Lock{
    
  private var mutex = pthread_mutex_t()
  private var attr = pthread_mutexattr_t()

  init() {
    pthread_mutexattr_init(&attr)
    pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE)
    pthread_mutex_init(&mutex, &attr)
  }

  deinit {
    pthread_mutexattr_destroy(&attr)
    pthread_mutex_destroy(&mutex)
  }

  func lock() {
    pthread_mutex_lock(&mutex)
  }

  func unlock() {
    pthread_mutex_unlock(&mutex)
  }
}

Task（任务）

Task这里我们的任务不需要做太多事情，所以我们直接选择闭包

typealias Task = () -> Void

taskQueue (任务队列)

taskQueue我选择用数组，当然你也可以自己构造队列

ThreadPriority

用来设置线程的优先级，一共分为三个等级，分别为high，normal，low

enum ThreadPriority {
    
    case high
    case normal
    case low
}

ThreadPool和DispatchQueue的具体结构之后会详细介绍，所以这里先不讲了。

ThreadPool

接口

Thread主要提供三个个方法：

  1. 初始化方法，可以通过初始化方法设置名称，线程池最大线程数和优先级。
  2. addTask方法，提交任务。
  3. workerThreadLoop方法，执行任务。

ThreadPool实现

通过最开始那个图我们应该大致能猜到需要哪些东西。首先，我们需要你一个锁用来控制任务执行，也就是我们创建的Condition，提供了和NSCondition一样的wait和sigal方法。

let lock = Condition()

为了确定何时启动新的线程，需要知道当前线程池中有多少线程，有多少线程正在运行，线程池最多支持多少线程:

var threadCount = 0  // 当前存在的线程个数
var activeThreadCount = 0 // 当前活跃的线程个数
var threadCountLimit = 0 // 线程池设置的最大线程个数

为了满足高级一点的需求，这里我们提供了设置线程优先级的功能，所以我们需要一个qos

let qos: qos_class_t  //用来设置线程优先级

最后就是任务队列，我们选择是的数组，因为我们的需求也很简单，进队，出队，删除，判空。

var tasksQueue = [DispatchTask]()  //用来存储任务

初始化很简单就不写了，现在重点讲一讲addTask方法，主要有几点需要注意：

1.为了保证tasksQueue的数据安全，在存入新的任务之前要先加锁

lock.lock()

tasksQueue.append(task)

2.我们是边添加任务边创建线程，这样可以创建合适的线程数，避免一次性创建过多不需要用的线程。phread创建线程有点麻烦

let idleThreads = threadCount - activeThreadCount
if tasksQueue.count > idleThreads && threadCount < threadCountLimit {
    
    let holder = Unmanaged.passRetained(self)
    let pointer = UnsafeMutableRawPointer(holder.toOpaque())
    var thread: pthread_t? = nil
    var user_interactive_qos_attr = pthread_attr_t()
    pthread_attr_init(&user_interactive_qos_attr)
    pthread_attr_set_qos_class_np(&user_interactive_qos_attr, qos, 0)
    
    if pthread_create(&thread, &user_interactive_qos_attr, threadRun, pointer) == 0 && thread != nil {
        holder.release()
    }
    threadCount += 1
}

private func threadRun(arg: UnsafeMutableRawPointer) -> UnsafeMutableRawPointer? {
    let unmanaged = Unmanaged<ThreadPool>.fromOpaque(arg)
    unmanaged.takeUnretainedValue().workerThreadLoop()
    unmanaged.release()
    return nil
}

workerThreadLoop是线程需要执行的方法。

接着我们需要唤醒线程开始来执行任务，至于为什么要唤醒，之后讲workerThreadLoop你就知道了。

lock.signal()

最后就是解锁了

lock.unlock()

下面我们来看看workerThreadLoop方法

fileprivate func workerThreadLoop() {
        
        if name != nil {
            pthread_setname_np(name!.cString(using: String.Encoding.utf8)!)
        }
        
        while true {
            
            autoreleasepool {
                let task = peekTask()
                    
                task()
                
                lock.lock()
                activeThreadCount -= 1
                lock.unlock()
            }
        }
    }

工作线程中主要是一个无限循环，这个和RunLoop类似。当tasksQueue为空时，线程进入休眠并等待。一旦数组不为空，该任务立即出队并开始执行。任务块执行开始后，当前活跃线程数加一；任务执行完成后，当前活跃线程数减一。这里我把任务出队列放在一个叫peekTask的方法中

private func peekTask() -> DispatchTask{
        
        lock.lock()

        while tasksQueue.isEmpty {
            lock.wait()
        }
        let task = tasksQueue.removeFirst()
        activeThreadCount += 1
        
        lock.unlock()
        
        return task
    }

其实我们整个ThreadPool设计是用的经典的生产者-消费者模式，当然也需要解决这个模式中的老大难的假性唤醒问题

什么是虚假唤醒？
举个例子，我们现在有一个生产者-消费者队列和三个线程。
I.1号线程从队列中获取了一个元素，此时队列变为空。
II.2号线程也想从队列中获取一个元素，但此时队列为空，2号线程便只能进入阻塞(cond.wait())，等待队列非空。
III.这时，3号线程将一个元素入队，并调用cond.notify()唤醒条件变量。
IV.处于等待状态的2号线程接收到3号线程的唤醒信号，便准备解除阻塞状态，执行接下来的任务(获取队列中的元素)。
V.然而可能出现这样的情况：当2号线程准备获得队列的锁，去获取队列中的元素时，此时1号线程刚好执行完之前的元素操作，返回再去请求队列中的元素，1号线程便获得队列的锁，检查到队列非空，就获取到了3号线程刚刚入队的元素，然后释放队列锁。 （即2号线程获得队列锁，却发现队列为空）
VI.等到2号线程获得队列锁，判断发现队列仍为空，1号线程“偷走了”这个元素，所以对于2号线程而言，这次唤醒就是“虚假”的，它需要再次等待队列非空。

防止虚假唤醒
cond.wait(locker, {return !q.empty(); });

我们这里防止虚假唤醒也是采用这种方式，也就是这段：

while tasksQueue.isEmpty {
    lock.wait()
}

下面我们再看看DispatchQueue吧

DispatchQueue

DispatchQueue接口

这我们模拟GCD中的DispatchQueue，一共提供了四个方法：

  1. 初始化方法，构造器，可通过它创建并行或串行队列。
  2. 异步派发方法async。
  3. 同步派发方法sync。

这里需要注意串行并行和同步异步，很多人容易搞混，不懂得可以看看这篇文章GCD中的串行/并行和同步/异步理解。

DispatchQueue实现

这部分我是参照Let's Build dispatch_queue实现的，然后在原有基础上添加了优先级，pthread可以通过qos_class类型值设置优先级，所以在初始化的时候需要做个优先级的类型转换。

let qos: qos_class_t = {
    switch priority {
    case .high:
        return QOS_CLASS_USER_INITIATED
    case .low:
        return QOS_CLASS_UTILITY
    case .normal:
        return QOS_CLASS_DEFAULT
    }
}()
    
threadPool = ThreadPool(name: name, threadCount: threadCout, qos: qos)

分发队列和线程池一样也需要锁，但这里的锁就相对简单些，不需要signal和wait，所以可以用我们构造的互斥锁Lock。

let lock = Lock()

分发队列需要依赖ThreadPool完成任务的执行，所以还需要一个线程池

let threadPool: ThreadPool

我们希望能提供设置队列是串行还是并行，如果是串行我们还需要记录是否有任务在执行

let serial: Bool  //判断是串行还是并行队列

var serialRunning = false

如果是串行队列我们就需要在DispatchQueue中暂存任务，等到上一个任务执行完成之后再将一下个任务添加到ThreadPool中，所以我们需要一个数组存储任务

var pendingTasks = [DispatchTask]()   //存储任务

在这里我把所有DispatchQueue的线程池的最大线程数都设置成128，这是一个随机数字，没有任何意义，你可以根据自己需要设置

let threadCout = 128

下面来看看异步派发方法async，异步方法简单些，任务不需要被等待，可以直接添加到线程池中，只需要告诉DispatchQueue有任务就行了。

同步派发方法就要复杂一点了，需要一个局部条件变量Condition和一个done来记录任务执行完成，获得锁任务未执行完成前condition.wait()，执行完成后condition.signal()并且condition.lock()

func sync(execute block: @escaping DispatchTask) {
        
    let condition = Condition()  //NSCondition 的对象实际上作为一个锁和一个线程检查器：锁主要为了当检测条件时保护数据源，执行条件引发的任务；线程检查器主要是根据条件决定是否继续运行线程，即线程是否被阻塞。
    var done = false
    
    self.async {
        block()
        condition.lock() //一般用于多线程同时访问、修改同一个数据源，保证在同一时间内数据源只被访问、修改一次，其他线程的命令需要在lock 外等待，只到unlock ，才可访问
        done = true
        condition.signal()  //CPU发信号告诉线程不用在等待，可以继续执行
        condition.unlock()
    }
    
    condition.lock()
    
    while !done {
        condition.wait()  //让当前线程处于等待状态
    }
    
    condition.unlock()
}

async和sync都是管理DispatchQueue中的任务执行，再来看看怎样将DispatchQueue中任务添加到ThreadPool中执行。

func dispatchOneTask() {
        
    threadPool.addTask {
        self.lock.lock()
        let block = self.pendingTasks.removeFirst()
        self.lock.unlock()
        
        block()
        
        if self.serial {
            self.lock.lock()
            if self.pendingTasks.count > 0 {
                self.dispatchOneTask()
            }else{
                self.serialRunning = false
            }
            self.lock.unlock()
        }
    }
}

并行队列很简单，只需要需要对pendingTasks加锁就行了。在串行队列上，新增的代码块需要等到前一个代码块执行完后才能执行。每当一个代码块执行完后，dispatchOneBlock 会检查当前队列是否还有代码块未执行。如果有的话，它会调用自身以便最后可以执行到该代码块。如果没有，将队列的运行状态设置回false。

好了，以上就是所有的内容，最后本文的demo在这PThreadPool

参考文章

Let's Build dispatch_queue
Pthread - 线程池（thread pool）实现
Strand
Aojet
Signals
Strand.swift

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,937评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,503评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,712评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,668评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,677评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,601评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,975评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,637评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,881评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,621评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,710评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,387评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,971评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,947评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,189评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,805评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,449评论 2赞 342

推荐阅读更多精彩内容

  • iOS多线程详解：实践篇
    iOS多线程实践中，常用的就是子线程执行耗时操作，然后回到主线程刷新UI。在iOS中每个进程启动后都会建立一个主线...
    jackyshan阅读 1,435评论 2赞 12
  • Java多线程（三）线程池
    为什么使用线程池 当我们在使用线程时，如果每次需要一个线程时都去创建一个线程，这样实现起来很简单，但是会有一个问题...
    闽越布衣阅读 4,276评论 10赞 45
  • iOS 面试宝典 没有比这更全的了（持续更新）
    1.ios高性能编程 (1).内层 最小的内层平均值和峰值(2).耗电量 高效的算法和数据结构(3).初始化时...
    欧辰_OSR阅读 29,299评论 8赞 265
  • 家庭成员
    回娘家，扳着手指算爸妈羽翼下的家庭成员，除了大大小小十九个娃儿，两条狗狗，一窝猫猫，竟然忘了，还有它们几十年不离不...
    Joydangqz阅读 674评论 0赞 0
  • 那日，闲
    朝游盘门东 幕至寒山南 不知何来此 恰逢与君识 知皆一面缘 又恐与君别 唯有心须静 才肯与君绝
    淋乂阅读 232评论 5赞 13