1.线程执行

join与setDaemon

• 1.子线程在主线程运行结束后，会继续执行完，如果给子线程设置为守护线程(setDaemon=True)，主线程运行结束子线程即结束；

• 2 .如果join()线程，那么主线程会等待子线程执行完再执行。

import threading
import time


def get_thread_a():
    print("get thread A started")
    time.sleep(3)
    print("get thread A end")


def get_thread_b():
    print("get thread B started")
    time.sleep(5)
    print("get thread B end")


if  __name__ == "__main__":
    thread_a = threading.Thread(target=get_thread_a)
    thread_b = threading.Thread(target=get_thread_b)
    start_time = time.time()
    thread_b.setDaemon(True)
    thread_a.start()
    thread_b.start()
    thread_a.join()   

    end_time = time.time()
    print("execution time: {}".format(end_time - start_time))

thread_a是join，首先子线程thread_a执行，thread_b是守护线程，当主线程执行完后，thread_b不会再执行

执行结果如下：

get thread A started
get thread B started
get thread A end
execution time: 3.003199815750122

2.线程同步

多线程间共享全局变量，多个线程对该变量执行不同的操作时，该变量最终的结果可能是不确定的(每次线程执行后的结果不同)，如：对count变量执行加减操作 ，count的值是不确定的，要想count的值是一个确定的需对线程执行的代码段加锁。

3.线程同步的方式

3.1 锁机制

python对线程加锁主要有Lock和Rlock模块

Lock：

from threading import Lock
 
lock = Lock()
lock.acquire()
lock.release()

Lock有acquire()和release()方法，这两个方法必须是成对出现的，acquire()后面必须release()后才能再acquire()，否则会造成死锁

Rlock：

鉴于Lock可能会造成死锁的情况，RLock(可重入锁)对Lock进行了改进，RLock可以在同一个线程里面连续调用多次acquire()，但必须再执行相同次数的release()

from threading import RLock

lock = RLock()
lock.acquire()
lock.acquire()
lock.release()
lock.release()

当一个线程调用锁的acquire()方法获得锁时，锁就进入“locked”状态。每次只有一个线程可以获得锁。如果此时另一个线程试图获得这个锁，该线程就会变为“blocked”状态，称为“同步阻塞”（参见多线程的基本概念）。

直到拥有锁的线程调用锁的release()方法释放锁之后，锁进入“unlocked”状态。线程调度程序从处于同步阻塞状态的线程中选择一个来获得锁，并使得该线程进入运行（running）状态。

3.2 Semaphore(信号量)

信号量也提供acquire方法和release方法，每当调用acquire方法的时候，如果内部计数器大于0，则将其减1，如果内部计数器等于0，则会阻塞该线程，直到有线程调用了release方法将内部计数器更新到大于1位置。

Semaphore（信号量）是计算机科学史上最古老的同步指令之一。Semaphore管理一个内置的计数器，每当调用acquire()时-1，调用release() 时+1。计数器不能小于0；当计数器为0时，acquire()将阻塞线程至同步锁定状态，直到其他线程调用release()。

基于这个特点，Semaphore经常用来同步一些有“访客上限”的对象，比如连接池。

BoundedSemaphore 与Semaphore的唯一区别在于前者将在调用release()时检查计数器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。

构造方法：
Semaphore(value=1): value是计数器的初始值。

import time
import threading


def get_thread_a(semaphore,i):
    time.sleep(1)
    print("get thread : {}".format(i))
    semaphore.release()


def get_thread_b(semaphore):
    for i in range(10):
        semaphore.acquire()
        thread_a = threading.Thread(target=get_thread_a, args=(semaphore,i))
        thread_a.start()


if __name__ == "__main__":
    semaphore = threading.Semaphore(2)
    thread_b = threading.Thread(target=get_thread_b, args=(semaphore,))
    thread_b.start()

3.3 条件判断

所谓条件变量，即这种机制是在满足了特定的条件后，线程才可以访问相关的数据。
  
它使用Condition类来完成，由于它也可以像锁机制那样用，所以它也有acquire方法和release方法，而且它还有wait，notify，notifyAll方法。

"""
一个简单的生产消费者模型，通过条件变量的控制产品数量的增减，调用一次生产者产品就是+1，调用一次消费者产品就会-1.
"""

"""
使用 Condition 类来完成，由于它也可以像锁机制那样用，所以它也有 acquire 方法和 release 方法，而且它还有
wait， notify， notifyAll 方法。
"""

import threading
import queue,time,random

class Goods:#产品类
    def __init__(self):
        self.count = 0
    def add(self,num = 1):
        self.count += num
    def sub(self):
        if self.count>=0:
            self.count -= 1
    def empty(self):
        return self.count <= 0

class Producer(threading.Thread):#生产者类
    def __init__(self,condition,goods,sleeptime = 1):#sleeptime=1
        threading.Thread.__init__(self)
        self.cond = condition
        self.goods = goods
        self.sleeptime = sleeptime
    def run(self):
        cond = self.cond
        goods = self.goods
        while True:
            cond.acquire()#锁住资源
            goods.add()
            print("产品数量:",goods.count,"生产者线程")
            cond.notifyAll()#唤醒所有等待的线程--》其实就是唤醒消费者进程
            cond.release()#解锁资源
            time.sleep(self.sleeptime)

class Consumer(threading.Thread):#消费者类
    def __init__(self,condition,goods,sleeptime = 2):#sleeptime=2
        threading.Thread.__init__(self)
        self.cond = condition
        self.goods = goods
        self.sleeptime = sleeptime
    def run(self):
        cond = self.cond
        goods = self.goods
        while True:
            time.sleep(self.sleeptime)
            cond.acquire()#锁住资源
            while goods.empty():#如无产品则让线程等待
                cond.wait()
            goods.sub()
            print("产品数量:",goods.count,"消费者线程")
            cond.release()#解锁资源

g = Goods()
c = threading.Condition()

pro = Producer(c,g)
pro.start()

con = Consumer(c,g)
con.start()

Condition内部有一把锁，默认是RLock，在调用wait()和notify()之前必须先调用acquire()获取这个锁，才能继续执行；当wait()和notify()执行完后，需调用release()释放这个锁，在执行with condition时，会先执行acquire()，with结束时，执行了release()；所以condition有两层锁，最底层锁在调用wait()时会释放，同时会加一把锁到等待队列，等待notify()唤醒释放锁

wait() ：允许等待某个条件变量的通知，notify()可唤醒

notify()： 唤醒等待队列wait()

# encoding: UTF-8
import threading
import time
 
# 商品
product = None
# 条件变量
con = threading.Condition()
 
# 生产者方法
def produce():
    global product
    
    if con.acquire():
        while True:
            if product is None:
                print 'produce...'
                product = 'anything'
                
                # 通知消费者，商品已经生产
                con.notify()
            
            # 等待通知
            con.wait()
            time.sleep(2)
 
# 消费者方法
def consume():
    global product
    
    if con.acquire():
        while True:
            if product is not None:
                print 'consume...'
                product = None
                
                # 通知生产者，商品已经没了
                con.notify()
            
            # 等待通知
            con.wait()
            time.sleep(2)
 
t1 = threading.Thread(target=produce)
t2 = threading.Thread(target=consume)
t2.start()
t1.start()

3.4 同步队列

put方法和task_done方法，queue有一个未完成任务数量num，put依次num+1，task依次num-1.任务都完成时任务结束。
  

import threading
import queue
import time
import random

'''
1.创建一个 Queue.Queue() 的实例，然后使用数据对它进行填充。
2.将经过填充数据的实例传递给线程类，后者是通过继承 threading.Thread 的方式创建的。
3.每次从队列中取出一个项目，并使用该线程中的数据和 run 方法以执行相应的工作。
4.在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号。
5.对队列执行 join 操作，实际上意味着等到队列为空，再退出主程序。
'''

class jdThread(threading.Thread):
    def __init__(self,index,queue):
        threading.Thread.__init__(self)
        self.index = index
        self.queue = queue

    def run(self):
        while True:
            time.sleep(1)
            item = self.queue.get()
            if item is None:
                break
            print("序号：",self.index,"任务",item,"完成")
            self.queue.task_done()#task_done方法使得未完成的任务数量-1

q = queue.Queue(0)
'''
初始化函数接受一个数字来作为该队列的容量，如果传递的是
一个小于等于0的数，那么默认会认为该队列的容量是无限的.
'''
for i in range(2):
    jdThread(i,q).start()#两个线程同时完成任务

for i in range(10):
    q.put(i)#put方法使得未完成的任务数量+1

3.5 Event对象

Event对象是一种简单的线程同步通信技术，一个线程设置Event对象，另一个线程等待Event对象。

import threading


# 自定义线程类
class MyThread(threading.Thread):
    def __init__(self, thread_name):
        threading.Thread.__init__(self, name=thread_name)

    # 重写线程代码
    def run(self):
        global my_event
        if my_event.isSet():
            my_event.clear()
            # 等待通知
            my_event.wait()
            print(self.getName())
        else:
            print(self.getName())
            my_event.set()


# 创建锁
my_event = threading.Event()
my_event.set()
t1 = []

for i in range(10):
    t = MyThread(str(i))
    t1.append(t)

for t in t1:
    t.start()

这里抛砖引玉，为何要使用同步？

当多个线程同时操作一个可共享的资源时会出现线程安全问题，将会导致数据不一致，因此使用同步锁来防止该操作执行完之前不许被其他线程执行，从而保证了该变量的唯一性和准确性。下面总结一些java线程实现同步方式，大致有下面几种：

1.同步方法 
    使用 synchronized关键字，可以修饰普通方法、静态方法，以及语句块。由于java的每个对象都有一个内置锁，当用此关键字修饰方法时， 内置锁会保护整个方法。在调用该方法前，需要获得内置锁，否则就处于阻塞状态。需要注意的是调用静态方法时，锁住的不是对象，锁住的是类。

 //修饰普通方法
 public synchronized void add(){}

 //修饰语句块
 synchronized(object){ 
    
 }
 //修饰静态方法
 public static synchronized int increase(){

 }

同步是一种高开销的操作，因此应该尽量减少同步的内容。 通常没有必要同步整个方法，使用synchronized代码块同步关键代码即可。 
      
2.使用特殊域变量(volatile)实现线程同步

• volatile关键字为域变量的访问提供了一种免锁机制， 
• 使用volatile修饰域相当于告诉虚拟机该域可能会被其他线程更新， 
• 因此每次使用该域就要重新计算，而不是使用寄存器中的值 
• volatile不会提供任何原子操作，它也不能用来修饰final类型的变量 

    多线程中的非同步问题主要出现在对域的读写上，如果让域自身避免这个问题，则就不需要修改操作该域的方法。 
    用final域，有锁保护的域和volatile域可以避免非同步的问题。 
    
3.使用重入锁实现线程同步

    在JavaSE5.0中新增了一个java.util.concurrent包来支持同步。 ReentrantLock类是可重入、互斥、实现了Lock接口的锁， 
    它与使用synchronized方法和快具有相同的基本行为和语义，并且扩展了其能力。
    ReenreantLock类的常用方法有：

•  ReentrantLock() : 创建一个ReentrantLock实例 
•  lock() : 获得锁 
•  unlock() : 释放锁 

 ReentrantLock()还有一个可以创建公平锁的构造方法，但由于能大幅度降低程序运行效率，不推荐使用 
   关于Lock对象和synchronized关键字的选择： 
        a.最好两个都不用，使用一种java.util.concurrent包提供的机制， 
            能够帮助用户处理所有与锁相关的代码。 
        b.如果synchronized关键字能满足用户的需求，就用synchronized，因为它能简化代码 
        c.如果需要更高级的功能，就用ReentrantLock类，此时要注意及时释放锁，否则会出现死锁，通常在finally代码释放锁 
        
4.使用局部变量实现线程同步 
       如果使用ThreadLocal管理变量，则每一个使用该变量的线程都获得该变量的副本， 副本之间相互独立，这样每一个线程都可以随意修改自己的变量副本，而不会对其他线程产生影响。

   ThreadLocal 类的常用方法

•     ThreadLocal() : 创建一个线程本地变量 
•     get() : 返回此线程局部变量的当前线程副本中的值 
•     initialValue() : 返回此线程局部变量的当前线程的"初始值" 
•     set(T value) : 将此线程局部变量的当前线程副本中的值设置为value

    ThreadLocal与同步机制 
        a.ThreadLocal与同步机制都是为了解决多线程中相同变量的访问冲突问题。 
        b.前者采用以"空间换时间"的方法，后者采用以"时间换空间"的方式

5.使用阻塞队列实现线程同步

上面的实现方式都是在底层实现的线程同步，但是我们在实际开发当中，应当尽量远离底层结构。 使用javaSE5.0版本中新增的java.util.concurrent包将有助于简化开发。 例如使用LinkedBlockingQueue<E>来实现线程的同步 。
    LinkedBlockingQueue<E>是一个基于已连接节点的，范围任意的blocking queue。 
    队列是先进先出的顺序（FIFO），关于队列以后会详细讲解~ 
    LinkedBlockingQueue 类常用方法 
    LinkedBlockingQueue() : 创建一个容量为Integer.MAX_VALUE的LinkedBlockingQueue 
    put(E e) : 在队尾添加一个元素，如果队列满则阻塞 
    size() : 返回队列中的元素个数 
    take() : 移除并返回队头元素，如果队列空则阻塞

BlockingQueue<E>定义了阻塞队列的常用方法，尤其是三种添加元素的方法，我们要多加注意，当队列满时：

• add()方法会抛出异常
• offer()方法返回false
• put()方法会阻塞

6.使用原子变量实现线程同步
     原子操作就是指将读取变量值、修改变量值、保存变量值看成一个整体来操作，即-这几种行为要么同时完成，要么都不完成。在java的util.concurrent.atomic包中提供了创建了原子类型变量的工具类，使用该类可以简化线程同步。其中AtomicInteger 表可以用原子方式更新int的值，可用在应用程序中(如以原子方式增加的计数器)，但不能用于替换Integer；可扩展Number，允许那些处理机遇数字类的工具和实用工具进行统一访问。
AtomicInteger类常用方法：

• AtomicInteger(int initialValue) : 创建具有给定初始值的新的AtomicInteger
• addAddGet(int dalta) : 以原子方式将给定值与当前值相加
• get() : 获取当前值

代码实例：

 1 class Bank {
 2         private AtomicInteger account = new AtomicInteger(100);
 3 
 4         public AtomicInteger getAccount() {
 5             return account;
 6         }
 7 
 8         public void save(int money) {
 9             account.addAndGet(money);
10         }
11 } 

补充--原子操作主要有：
　　对于引用变量和大多数原始变量(long和double除外)的读写操作；
　　对于所有使用volatile修饰的变量(包括long和double)的读写操作。

Overview

C++的多线程同步方式有这么几种：

• mutex
  • lock_guard
  • unique_lock
• condition_variable
• future
  • promise
  • packaged_task
  • async

C++11并没有提供semaphore的API，信号量太容易出错了（too error prone），通过组合互斥锁（mutex）和条件变量（condition variable）可以达到相同的效果，且更加安全。

mutex

官网介绍：mutex
它包含以下三个部分：

• Mutex type
• Locks：lock_guard， unique_lock
• Functions：try_lock，lock

example：
mutex的实现也很简单，在进入临界区之前调用该变量（mtx）的lock函数，出临界区之前调用该变量的（mtx）的unlock函数。所以程序会连续输出50个'*'或者连续输出50个'$'，而不会'*' '$'交替输出。

但是考虑这样一个问题，std::cout << '\n'，这里发生异常会发生什么？抛出异常后，意味着mtx.unlock()不会被执行，即锁没有被释放，整个程序进入不了临界区，该程序往往会挂死。

// mutex example
#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex

std::mutex mtx;           // mutex for critical section

void print_block (int n, char c) {
  // critical section (exclusive access to std::cout signaled by locking mtx):
  mtx.lock();
  for (int i=0; i<n; ++i) { std::cout << c; }
  std::cout << '\n';
  mtx.unlock();
}

int main ()
{
  std::thread th1 (print_block,50,'*');
  std::thread th2 (print_block,50,'$');

  th1.join();
  th2.join();

  return 0;
}

Possible output (order of lines may vary, but characters are never mixed):

**************************************************
$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$

lock_guard

在构造时，互斥对象被调用线程锁定，而在销毁时，互斥对象被解锁。它是最简单的锁，作为具有自动持续时间的对象特别有用，该持续时间一直持续到其上下文结束。这样，可以保证在抛出异常的情况下互斥对象已正确解锁。

example

// lock_guard example
#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex, std::lock_guard
#include <stdexcept>      // std::logic_error

std::mutex mtx;

void print_even (int x) {
  if (x%2==0) std::cout << x << " is even\n";
  else throw (std::logic_error("not even"));
}

void print_thread_id (int id) {
  try {
    // using a local lock_guard to lock mtx guarantees unlocking on destruction / exception:
    std::lock_guard<std::mutex> lck (mtx);
    print_even(id);
  }
  catch (std::logic_error&) {
    std::cout << "[exception caught]\n";
  }
}

int main ()
{
  std::thread threads[10];
  // spawn 10 threads:
  for (int i=0; i<10; ++i)
    threads[i] = std::thread(print_thread_id,i+1);

  for (auto& th : threads) th.join();

  return 0;
}

possible output

[exception caught]
2 is even
[exception caught]
4 is even
[exception caught]
6 is even
[exception caught]
8 is even
[exception caught]
10 is even

unique_lock

unique_lock基本用法和lock_guard一致，在构造函数和析构函数中进行锁操作，不同的地方在于它提供了非常多构造函数。

example

// unique_lock example
#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex, std::unique_lock

std::mutex mtx;           // mutex for critical section

void print_block (int n, char c) {
  // critical section (exclusive access to std::cout signaled by lifetime of lck):
  std::unique_lock<std::mutex> lck (mtx);
  for (int i=0; i<n; ++i) { std::cout << c; }
  std::cout << '\n';
}

int main ()
{
  std::thread th1 (print_block,50,'*');
  std::thread th2 (print_block,50,'$');

  th1.join();
  th2.join();

  return 0;
}

condition_variable

条件变量是一个对象，可以阻塞线程，直到被通知恢复。当调用其等待功能之一时，它使用unique_lock（通过互斥锁）来锁定线程。该线程将保持阻塞状态，直到被另一个在同一个condition_variable对象上调用通知功能的线程唤醒为止。
Wait functions

• wait
  • Wait until notified (public member function )
• wait_for
  • Wait for timeout or until notified (public member function )
• wait_until
  • Wait until notified or time point (public member function )

Notify functions

• notify_one
  • Notify one (public member function )
• notify_all
  • Notify all (public member function )

example

// condition_variable::notify_one
#include <iostream>           // std::cout
#include <thread>             // std::thread
#include <mutex>              // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variable

std::mutex mtx;
std::condition_variable produce,consume;

int cargo = 0;     // shared value by producers and consumers

void consumer () {
  std::unique_lock<std::mutex> lck(mtx);
  while (cargo==0) consume.wait(lck);
  std::cout << cargo << '\n';
  cargo=0;
  produce.notify_one();
}

void producer (int id) {
  std::unique_lock<std::mutex> lck(mtx);
  while (cargo!=0) produce.wait(lck);
  cargo = id;
  consume.notify_one();
}

int main ()
{
  std::thread consumers[10],producers[10];
  // spawn 10 consumers and 10 producers:
  for (int i=0; i<10; ++i) {
    consumers[i] = std::thread(consumer);
    producers[i] = std::thread(producer,i+1);
  }

  // join them back:
  for (int i=0; i<10; ++i) {
    producers[i].join();
    consumers[i].join();
  }

  return 0;
}

Possible output (order of consumed cargoes may vary):

1
2
3
4
5
6
7
8
9
10

future

future的目标是充分利用CPU的并发性，它只能通过async，promise和package_task三种方式构造。future只能移动，不可复制，需要复制时可以使用shared_future，但通常不建议使用。调用future的get()时可能会发生阻塞，直到返回值ready。future有三种姿势的等待：

• wait()：一直等待直到得到返回值
• wait_for()：设定一个超时时间；
• wait_until()：等待到某个时间点。

future有一特化版本future，返回值为空，即不返回任何值，因此仅能用于线程间通知，但却是最常用的future。

promise

promise对象可以通过调用成员get_future将此共享状态与future对象关联。调用之后，两个对象共享相同的共享状态：

• promise：promise对象是异步提供者，在共享状态的时候设置一个值
• future负责：future对象是异步返回对象，可以检索共享状态的值，并在必要时等待其准备就绪

example

// promise example
#include <iostream>       // std::cout
#include <functional>     // std::ref
#include <thread>         // std::thread
#include <future>         // std::promise, std::future

void print_int (std::future<int>& fut) {
  int x = fut.get();
  std::cout << "value: " << x << '\n';
}

int main ()
{
  std::promise<int> prom;                      // create promise

  std::future<int> fut = prom.get_future();    // engagement with future

  std::thread th1 (print_int, std::ref(fut));  // send future to new thread

  prom.set_value (10);                         // fulfill promise
                                               // (synchronizes with getting the future)
  th1.join();
  return 0;
}

Output:

value: 10

packaged_task

很多情况下并不希望另起一个线程，因为线程是非常重要的资源。因此希望可以合理的管理线程资源，这就需要使用线程池。如何将future与线程池同时使用呢？这就需要采用package_task。package_task本质是将一个函数包装成一个future。这个task类似于std::function，有输入输出，大家可以将其认为是一个异步函数，但该异步函数并不负责执行，而是将其结果预置于一个future变量中，然后交给一个线程来实际执行，此时主线程便可以得到其返回值。

example

// packaged_task example
#include <iostream>     // std::cout
#include <future>       // std::packaged_task, std::future
#include <chrono>       // std::chrono::seconds
#include <thread>       // std::thread, std::this_thread::sleep_for

// count down taking a second for each value:
int countdown (int from, int to) {
  for (int i=from; i!=to; --i) {
    std::cout << i << '\n';
    std::this_thread::sleep_for(std::chrono::seconds(1));
  }
  std::cout << "Lift off!\n";
  return from-to;
}

int main ()
{
  std::packaged_task<int(int,int)> tsk (countdown);   // set up packaged_task
  std::future<int> ret = tsk.get_future();            // get future

  std::thread th (std::move(tsk),10,0);   // spawn thread to count down from 10 to 0

  // ...

  int value = ret.get();                  // wait for the task to finish and get result

  std::cout << "The countdown lasted for " << value << " seconds.\n";

  th.join();

  return 0;
}

Possible output:

10
9
8
7
6
5
4
3
2
1
Lift off!
The countdown lasted for 10 seconds.

async

有时某项工作很早就可以开始做（前置条件都已完备），而等待这件工作结果的任务在非常靠后的位置，这时候就需要async。换言之，如果可以尽早开始做一件事，就让其在后台运行即可，或快或慢都可以，只需在需要结果的时候运行完成就好。

example

// async example
#include <iostream>       // std::cout
#include <future>         // std::async, std::future

// a non-optimized way of checking for prime numbers:
bool is_prime (int x) {
  std::cout << "Calculating. Please, wait...\n";
  for (int i=2; i<x; ++i) if (x%i==0) return false;
  return true;
}

int main ()
{
  // call is_prime(313222313) asynchronously:
  std::future<bool> fut = std::async (is_prime,313222313);

  std::cout << "Checking whether 313222313 is prime.\n";
  // ...

  bool ret = fut.get();      // waits for is_prime to return

  if (ret) std::cout << "It is prime!\n";
  else std::cout << "It is not prime.\n";

  return 0;
}

Possible output (the first two lines may be in a different order, or scrambled):

Checking whether 313222313 is prime.
Calculating. Please, wait...
It is prime!

Reference

10分钟，带你掌握C++多线程同步！
Multi-threading