python并发編程之多线程

并发与并行

cup是顺序执行的，操作系统轮流让各个任务交替执行，每个任务在时间片内执行，时间片结束切换到下一个任务，这是并发。
真正的并行执行多任务只能在 多核CPU 上实现，但是，由于任务数量远远多于 CPU 的核心数量，所以，操作系统也会自动把很多任务轮流调度到每个核心上执行。

并发性：是指两个或多个事件在同一时间间隔内发生。
并行性：是指两个或多个事件在同一时刻发生。

举个例子：我们有一个车间（cpu），可以生产各种东西（任务），比如汽车，飞机，自行车。制造汽车时：车间每生产一个零件（计算操作）后，需要等它冷却，定型，组装好后才能生产下一个零件，在这个冷却，定型，组装的过程（IO操作）中车间是不工作的（程序霸占着cpu时间，效率低）。为了让我们车间效率高，我们让它在冷却，定型，组装的时间中生产其他的零件，等组装好后再继续生产汽车零件。某个确定的时间时做一个事情，但在一个时间段（时间片）内做多个事情，这就是并发。多个车间同时工作就是并行。

【拓展：时间片】
每个进程被分配一个时间段，称作它的时间片，即该进程允许运行的时间。如果在时间片结束时进程还在运行，则CPU将被剥夺并分配给另一个进程。如果进程在时间片结束前阻塞或结束，则CPU当即进行切换。调度程序所要做的就是维护一张就绪进程列表，当进程用完它的时间片后，它被移到队列的末尾。

从一个进程切换到另一个进程是需要一定时间的--保存和装入寄存器值及内存映像，更新各种表格和队列等。
时间片设得太短会导致过多的进程切换，降低了CPU效率；而设得太长又可能引起对短的交互请求的响应变差。将时间片设为100毫秒通常是一个比较合理的折中。

进程和线程

进程

每个独立的线程有一个程序运行的入口、顺序执行序列和程序的出口。但是线程不能够独立执行，必须依存在应用程序中，由应用程序提供多个线程执行控制。
每个线程都有他自己的一组CPU寄存器，称为线程的上下文，该上下文反映了线程上次运行该线程的CPU寄存器的状态。
指令指针和堆栈指针寄存器是线程上下文中两个最重要的寄存器，线程总是在进程得到上下文中运行的，这些地址都用于标志拥有线程的进程地址空间中的内存。

• 线程可以被抢占（中断）。
• 在其他线程正在运行时，线程可以暂时搁置（也称为睡眠） -- 这就是线程的退让。

线程可以分为:

• 内核线程：由操作系统内核创建和撤销。
• 用户线程：不需要内核支持而在用户程序中实现的线程。

进程和程序的关系

多线程

为什么要使用多线程？
    线程在程序中是独立的、并发的执行流。与分隔的进程相比，进程中线程之间的隔离程度要小，它们共享内存、文件句柄
    和其他进程应有的状态。
    因为线程的划分尺度小于进程，使得多线程程序的并发性高。进程在执行过程之中拥有独立的内存单元，而多个线程共享
    内存，从而极大的提升了程序的运行效率。
    线程比进程具有更高的性能，这是由于同一个进程中的线程都有共性，多个线程共享一个进程的虚拟空间。线程的共享环境
    包括进程代码段、进程的共有数据等，利用这些共享的数据，线程之间很容易实现通信。
    操作系统在创建进程时，必须为改进程分配独立的内存空间，并分配大量的相关资源，但创建线程则简单得多。因此，使用多线程
    来实现并发比使用多进程的性能高得要多。

总结起来，使用多线程编程具有如下几个优点：
    进程之间不能共享内存，但线程之间共享内存非常容易。
    操作系统在创建进程时，需要为该进程重新分配系统资源，但创建线程的代价则小得多。因此使用多线程来实现多任务并发执行比使用多进程的效率高

python的多线程

python语言内置了多线程功能支持，而不是单纯地作为底层操作系统的调度方式，从而简化了python的多线程编程。
在 CPython 中，由于存在 全局解释器锁，同一时刻只有一个线程可以执行 Python 代码，如果你想让你的应用更好地利用多核心计算机的计算资源，推荐你使用 multiprocessing 或 concurrent.futures.ProcessPoolExecutor。 但是，如果你想要同时运行多个 I/O 密集型任务，则多线程仍然是一个合适的模型。

python实现多线程

• _thread
• threading(推荐使用)

thread 模块已被废弃。用户可以使用 threading 模块代替。所以，在 Python3 中不能再使用"thread" 模块。为了兼容性，Python3 将 thread 重命名为 "_thread"。
这里只讲threading。

threading模块

简单示例：

import threading

# 最简单的线程程序
def worker():
    print("I'm working")
    print('Fineshed')

t = threading.Thread(target=worker, name='worker') # 线程对象
t.start() #  启动

threading.Thread参数解析

• target    线程调用的对象，就是目标函数
• name    为线程起个名字
• args    为目标函数传递实参，元组
• kwargs    为目标函数关键字传参，字典

通过threading.Thread创建一个线程对象，  target是目标函数，可以使用name为线程指定名称。 但是线程没有启动，需要调用start方法。
线程之所以执行函数，是因为线程中就是要执行代码的，而最简单的代码封装就是函数，所以还是函数 调用。
函数执行完，线程也就退出了。

线程退出
Python没有提供线程退出的方法，线程在下面情况时退出

• 线程函数内语句执行完毕
• 线程函数中抛出未处理的异常

# 程序抛异常线程退出
import threading
import time

def worker():
    for i in range(10):
        time.sleep(0.5)
        if i > 5:
        #break # 终止循环
        #return # 函数返回
            raise RuntimeError # 抛异常
    print('I am working')
    print('finished')

t = threading.Thread(target=worker, name='worker')
t.start()

print('=' * 30)

这个模块定义了以下函数：

• threading.main_thread()  
  返回主 Thread 对象。一般情况下，主线程是Python解释器开始时创建的线程。
  
• threading.current_thread():
  返回当前对应调用者的控制线程的 Thread 对象。如果调用者的控制线程不是利用 threading 创建，会返回一个功能受限的虚拟线程对象。
• threading.enumerate():
  返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。

• threading.active_count():
  返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
• threading.get_ident():
  返回当前线程的 “线程标识符”。它是一个非零的整数。它的值没有直接含义，主要是用作 magic cookie，比如作为含有线程相关数据的字典的索引。线程标识符可能会在线程退出，新线程创建时被复用。
• threading.excepthook(args, /)
  处理由 Thread.run() 引发的未捕获异常。
• ...

import threading
import time

def showtreadinfo():
    # 打印当前线程对象，主线程对象，活跃线程数
    print('current thread = {}\nmain thread = {}\nactive count = {}' .format(
    threading.current_thread(), threading.main_thread(), threading.active_count()))
    # 打印正在运行的线程的list，线程标识符，
    print('threading.enumerate = {}\nthreading.get_ident = {}'.format(
    threading.enumerate(),threading.get_ident()))


def worker():
    showtreadinfo()
    for i in range(2):
        time.sleep(1)
        print('i am working')
        print('finished')

t = threading.Thread(target=worker, name='worker') # 线程对象
showtreadinfo()
time.sleep(1)
t.start() # 启动
print('===end===')

# current thread = <_MainThread(MainThread, started 17200)>
# main thread = <_MainThread(MainThread, started 17200)>
# active count = 1
# threading.enumerate = [<_MainThread(MainThread, started 17200)>]
# threading.get_ident = 17200
# current thread = <Thread(worker, started 10784)>
# main thread = <_MainThread(MainThread, started 17200)>
# active count = 2
# threading.enumerate = [<_MainThread(MainThread, started 17200)>, <Thread(worker, started 10784)>]
# threading.get_ident = 10784
# ===end===
# i am working
# finished
# i am working
# finished

线程对象

除了使用方法外，线程模块同样提供了Thread类来处理线程

threading.Thread()

创建一个daemon=True的守护线程：

import time
import threading

def foo():
    time.sleep(1)
    for i in range(2):
        print(i)

# 主线程是non-daemon线程
t = threading.Thread(target=foo, daemon=True)   # 主线程执行完直接退出，不会等待t结束
t.start()

print('Main Thread Exits')

# Main Thread Exits

主线程是non-daemon线程，即daemon = False。

• 线程具有一个daemon属性，可以手动设置为True或False，也可以不设置，则取默认值None 如果不设置daemon，就取当前线程的daemon来设置它
• 主线程是non-daemon线程，即daemon = False
• 从主线程创建的所有线程的不设置daemon属性，则默认都是daemon = False，也就是non- daemon线程
• Python程序在没有活着的non-daemon线程运行时，程序退出，也就是除主线程之外剩下的只能 都是daemon线程，主线程才能退出，否则主线程就只能等待

Thread类提供了以下属性和方法:

• name    只是一个名字，只是个标识，名称可以重名。  getName()、  setName()获取、设置这 个名词
• ident    线程ID，它是非0整数。线程启动后才会有ID，否则为None。线程退出，此ID依旧可 以访问。此ID可以重复使用
  
• native_id    此线程的线程 ID (TID)，由 OS (内核) 分配。 这是一个非负整数，或者如果线程还未启动则为 None。
• daemon  一个表示这个线程是（True）否（False）守护线程的布尔值。一定要在调用 start() 前设置好，不然会抛出 RuntimeError 。初始值继承于创建线程；主线程不是守护线程，因此主线程创建的所有线程默认都是 daemon = False。当没有存活的非守护线程时，整个Python程序才会退出。

  
  

  
  

  
  

• start():启动线程活动。start方法才能启动操作系统线程，并运行run方法。  run方法内部调用了目标函数。
• run(): 用以表示线程活动的方法。
• join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。
• is_alive(): 返回线程是否活动的。
• getName(): 返回线程名。
• setName(): 设置线程名。
• isDaemon()    返回线程是否是守护线程
• setDaemon() 设置daemon值

创建多个线程：

import threading
import time
import sys

def worker(f=sys.stdout):
    t = threading.current_thread()
    for i in range(2):
        time.sleep(1)
        print('i am working' , t.name, t.ident, file=f)
        print('finished', file=f)

t1 = threading.Thread(target=worker, name='worker1')
t2 = threading.Thread(target=worker, name='worker2', args=(sys.stderr,))
t1.start()
t2.start()

# i am working worker1 18412
# finished
# i am working worker2 1312
# finished
# i am working worker2 1312
# finished
# i am working worker1 18412
# finished

可以看到worker1和work2交替执行。

当使用start方法启动线程后，进程内有多个活动的线程并行的工作，就是多线程。
一个进程中至少有一个线程，并作为程序的入口，这个线程就是主线程。
一个进程至少有一个主线程。其他线程称为工作线程。

线程安全

多线程执行一段代码，不会产生不确定的结果，那这段代码就是线程安全的。

多线程在运行过程中，由于共享同一进程中的数据，多线程并发使用同一个数据，那么数据就有可能被 相互修改，从而导致某些时刻无法确定这个数据的值，最终随着多线程运行，运行结果不可预期，这就 是线程不安全。

import threading
import time
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

class A:
    def __init__(self):
        self.x = 0

# 全局对象
global_data = A()

def worker():
    global_data .x = 0
    for i in range(100):
        time.sleep(0.0001)
        global_data .x += 1
        logging.info(global_data.x)


for i in range(10):
     threading.Thread(target=worker, name='t-{}'.format(i)).start()

# 每次执行global_data .x的值都是不确定的

threading.local类

import threading
import time
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)
# 全局对象
global_data = threading.local()

def worker():
    global_data .x = 0
    for i in range(100):
        time.sleep(0.0001)
        global_data .x += 1
        logging.info(global_data.x)

for i in range(10):
    threading.Thread(target=worker, name='t-{}'.format(i)).start()

# 当前线程创建的threading.local()的属性被其他线程访问时会报错

import logging
import threading
import time

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)
# 全局对象
X = 'abc'
global_data = threading.local()
global_data .x = 100
print(global_data , global_data .x)
print('~' * 30)
time.sleep(2)

def worker():
    logging.info(X)
    logging.info(global_data)
    logging.info(global_data.x)

worker() # 普通函数调用
print('=' * 30)
time.sleep(2)
threading.Thread(target=worker, name='worker').start() # 启动一个线程

# AttributeError: '_thread._local' object has no attribute 'x'

{ id(Thread) -> (ref(Thread), thread-local dict) } 

# 老板监视工人生产10个杯子

from threading import Event, Thread
import logging
import time
FORMAT = '%(asctime)s %(threadName)s %(thread)s %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

def boss(event:Event):
    logging.info("I'm boss, waiting for U")
    event.wait() # 阻塞等待
    logging.info('Good Job.')

def worker(event:Event, count=10):
    logging.info('I am working for U')
    cups = []
    while not event.wait(0.1):
        logging.info('make 1 cup')
        time.sleep(0.1)
        cups.append(1)
        if len(cups) >= count:
            event.set()
    logging.info('I finished my job. cups={}' .format(cups))

event = Event()
b = Thread(target=boss, name='boss', args=(event,))
w = Thread(target=worker, name='worker', args=(event,))
b.start()
w.start()

import threading
from threading import Thread, Lock
import time
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)
cups = []
lock = Lock() # 锁

def worker(count=100):
    logging.info("I'm working")
    while True:
        # 生产杯子期间加锁（改变cups列表）
        lock.acquire() # 获取锁

        if len(cups) >= count:
            lock.release()
            break

        time.sleep(0.0001) # 为了看出线程切换效果，模拟杯子制作时间
        cups.append(1)
        lock.release()
    logging.info('I finished my job. cups = {}' .format(len(cups)))

for i in range(1, 11):
    t = Thread(target=worker, name="w{}".format(i), args=(1000,))
    t.start()

import threading
from threading import Thread, Lock
import time
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)
cups = []
lock = Lock() # 锁

def worker(count=100):
    logging.info("I'm working")
    while True:
        with lock: # 获取锁，离开with释放锁
            if len(cups) >= count:
                logging.info('leaving')
                break
            time.sleep(0.0001) # 为了看出线程切换效果，模拟杯子制作时间
            cups.append(1)
            logging.info(lock.locked())

    logging.info('I finished my job. cups = {}' .format(len(cups)))

for i in range(1, 11):
    t = Thread(target=worker, name="w{}".format(i), args=(1000,))
    t.start()

# 递归锁：RLcok类的用法和Lock类一模一样，但它支持嵌套，在多个锁没有释放的时候一般会使用RLock类
import threading
import time

def func(lock):
    global gl_num
    lock.acquire()
    gl_num += 1
    time.sleep(1)
    print(gl_num)
    lock.release()


if __name__ == '__main__':
    gl_num = 0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=func,args=(lock,))
        t.start()

maxconnections = 5
# ...
pool_sema = BoundedSemaphore(value=maxconnections)

with pool_sema:
    conn = connectdb()
    try:
        # ... use connection ...
    finally:
        conn.close()

from threading import Thread,Semaphore
import threading
import time
# def func():
#     if sm.acquire():
#         print (threading.currentThread().getName() + ' get semaphore')
#         time.sleep(2)
#         sm.release()
def func():
    sm.acquire()
    print('%s get sm' %threading.current_thread().getName())
    time.sleep(3)
    sm.release()
if __name__ == '__main__':
    sm=Semaphore(5)
    for i in range(23):
        t=Thread(target=func)
        t.start()

import threading

def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()
def condition_func():

    ret = False
    inp = input('>>>')
    if inp == '1':
        ret = True

    return ret


def run(n):
    con.acquire()
    con.wait_for(condition_func)
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()