python-线程

★概念

进程是由若干线程组成的，一个进程至少有一个线程；

线程是CPU调度的最小单位；

线程之间资源共享。

★全局解释器锁（GIL）

cpython的特性

同一时刻只能有一个线程访问CPU；

锁的是线程

在多线程环境中，Python 虚拟机按以下方式执行：

1，设置GIL；　　

2，切换到一个线程去执行；　　

3，运行指定数量字节码指令或线程主动让出控制；　　

4，把线程设置为睡眠状态；　　

5，解锁 GIL；　　

6，再次重复以上所有步骤。　　　　　　

★threading模块

◇线程的创建-threading.Thread

import time
import os
from threading import Thread

"""多线程并发"""


def func(n):
    time.sleep(1)
    print('in func%s' % n)
    print('子进程PID：%s' % os.getpid())


for i in range(10):
    t = Thread(target=func, args=(i+1,))    # 并发10个线程
    t.start()

print('主进程PID：%s' % os.getpid())

import time
from threading import Thread


class MyThread(Thread):     # 继承线程类
    def __init__(self, arg):    # 初始化MyThread的参数
        super().__init__()
        self.arg = arg

    def run(self):      # 重写run方法
        time.sleep(1)
        print('in func%s' % self.arg)


for i in range(10):
    t = MyThread(i+1)
    t.start()

import os
from threading import Thread


def func(a, b):
    global g    # 声明全局变量，主线程和子线程共享g
    c = a + b
    g = 0       # 这个g同时对主线程的g有作用
    print('in func的结果:%s' % c)


g = 100
t_lis = []
for i in range(10):
    t = Thread(target=func, args=(i+1, 5))
    t.start()
    t_lis.append(t)

for t in t_lis:
    t.join()

print('主线程g的值：', g)        # g的值已被子线程修改
print('主进程PID：%s' % os.getpid())

守护线程

守护线程会在主线程代码结束之后等待【其他子线程】的结束而结束

与进程相同：

　　△主线程会等待子线程结束才结束

　　△主进程会等待子进程结束才结束

与进程不同：　　

　　▽守护线程在【其他线程结束】才结束　　-- >这时候主线程已经结束

　　▽守护进程在【代码结束】而结束 　　　　--> 这时候主进程还没结束

from threading import Thread
import time


def func1():
    while True:
        print('**********')
        time.sleep(1)


def func2():
    time.sleep(10)
    print('in func2')


t = Thread(target=func1,)
t.daemon = True     # 设置守护线程， 会一直等所有子线程执行结束，守护线程才结束
t.start()
t1 = Thread(target=func2,)
t1.start()
# t1.join()
print('in main thread')

◇线程的锁-threading.Lock

方法

acquire()　　上锁

release()　　解锁

Lock()  互斥锁

from threading import Lock

lock = Lock()

lock.acquire()
# lock.release()    
lock.acquire()
print('ok')

互斥锁（英语：Mutual exclusion，缩写 Mutex）是一种用于多线程编程中，防止两条线程同时对同一公共资源（比如全局变量）进行读写的机制。该目的通过将代码切片成一个一个的临界区域（critical section）达成。临界区域指的是一块对公共资源进行访问的代码，并非一种机制或是算法。一个程序、进程、线程可以拥有多个临界区域，但是并不一定会应用互斥锁。

　　有两个互斥锁时，会有可能产生死锁问题

from threading import Thread, Lock
import time

"""互斥锁"""


def func(lock):
    lock.acquire()  # 一个线程执行完了，下一个线程才能进去执行
    global n
    temp = n
    time.sleep(0.2)
    n = temp - 1
    lock.release()


n = 10
lock = Lock()
t_lst = []
for i in range(10):
    t = Thread(target=func, args=(lock,))
    t_lst.append(t)
    t.start()

for t in t_lst:
    t.join()
print(n)

　　ps:如果加锁，1，第一个线程执行，遇到time.sleep阻塞，时间片不切换，等待阻塞完成，执行完这个线程，2，（这时候第一个执行的线程执行完）CPU执行下一个线程，3，再下一个线程重复以上操作

　　ps:如果没有锁，1，第一个线程执行，遇到time.sleep阻塞；2，时间片切换，CPU执行下一个线程（这时候第一个执行的线程还没有执行完），3，再下一个线程重复以上操作  -->大家共享阻塞的时间（相当于同时等待）

from threading import Thread, Lock
import time


def eat1(name):
    noodle_lock.acquire()
    print('%s拿到了面条' % name)
    fork_lock.acquire()
    print('%s拿到了叉子' % name)
    print('%s 吃面了' % name)
    noodle_lock.release()
    fork_lock.release()


def eat2(name):
    fork_lock.acquire()
    print('%s拿到了叉子' % name)
    time.sleep(0.1)     # 先阻塞一下（迫使时间片切换），让另一个线程去拿了面条的锁，然后就会产生死锁的问题
    noodle_lock.acquire()   # 其他线程拿了锁没有解锁，导致我这里无法拿到而一直等待，
    print('%s拿到了面条' % name)
    print('%s 吃面了' % name)
    noodle_lock.release()
    fork_lock.release()


noodle_lock = Lock()
fork_lock = Lock()
Thread(target=eat1, args=('alex', )).start()
Thread(target=eat2, args=('jin', )).start()
Thread(target=eat1, args=('boss', )).start()
Thread(target=eat2, args=('egg', )).start()

RLock()　　递归锁

　　1，在一个线程能有多个锁

　　2，为了解决死锁问题

from threading import RLock

rlock = RLock()

rlock.acquire()
rlock.acquire()
rlock.acquire()
print('hello')

from threading import Thread, RLock
import time


def eat1(name):
    noodle_lock.acquire()
    print('%s拿到了面条' % name)
    fork_lock.acquire()
    print('%s拿到了叉子' % name)
    print('%s 吃面了' % name)
    noodle_lock.release()
    fork_lock.release()


def eat2(name):
    fork_lock.acquire()     # 上锁了
    print('%s拿到了叉子' % name)
    time.sleep(2)           # 阻塞，就算切换到其他线程也无法执行，因为是同一把锁，另外的线程外面也有锁
    noodle_lock.acquire()
    print('%s拿到了面条' % name)
    print('%s 吃面了' % name)
    noodle_lock.release()
    fork_lock.release()     # 已经完全解锁了，其他线程能去上锁了然后去执行


noodle_lock = fork_lock = RLock()
Thread(target=eat1, args=('alex', )).start()
Thread(target=eat2, args=('jin', )).start()
Thread(target=eat1, args=('boss', )).start()
Thread(target=eat2, args=('egg', )).start()

◇信号量-threading.Semaphore

方法

acquire()　　上锁

release()　　解锁

同一时间只有n个线程可以执行上锁代码

from threading import Semaphore, Thread
import time


def func(n):
    sem.acquire()
    time.sleep(1)
    print(n)
    sem.release()


sem = Semaphore(3)      # 设置一个信号量， 只能有3个线程可以并发执行
for i in range(10):
    t = Thread(target=func, args=(i,))      # 在线程里sem不用传进来，子线程共享主线程的变量
    t.start()

◇事件-threading.Event

状态

　　False　　（默认）　　wait()阻塞

　　True　　　　　　　　wait()非阻塞

修改状态

　　set　　设置为True

　　clear　　设置为False

wait方法

　　wait(1)　　状态为False的时候，只等待1s就往下执行

from threading import Event, Thread
import time
import random


def connect_db():
    count = 0
    while count < 3:
        e.wait(1)   # 只阻塞1S，往下执行
        if e.is_set():
            print('连接数据库成功！！！')
            break
        else:
            count += 1
            print('第%s次连接失败' % count)
    else:
        raise TimeoutError('连接超时!!!')


def check_web():
    time.sleep(random.randint(0, 3))    # 模拟网络延迟
    e.set()


e = Event()
t1 = Thread(target=connect_db)      # 直到事件状态被设置为True，这个线程才能连接上数据库
t2 = Thread(target=check_web)      # 这个线程负责修改事件状态
t1.start()
t2.start()

◇条件-threading.Condition

使得线程等待，只有满足某条件时，才释放n个线程

方法

　　acquire()　　

　　release()　　　　　　

　　notify() 造钥匙（一次性钥匙）　　　　

　　wait() 等钥匙　　

from threading import Condition, Thread


def func(i):
    con.acquire()
    con.wait()          # 等造钥匙出来
    print('在%s循环里' % i)
    con.release()


con = Condition()
for i in range(10):
    t = Thread(target=func, args=(i, ))
    t.start()
while True:
    num = int(input('>>>'))
    con.acquire()
    con.notify(num)     # 造钥匙（一次性钥匙）
    con.release()

◇定时器-threading.Timer

指定n秒后执行某个线程

from threading import Timer
import time


def func():
    print('时间同步')


while True:
    Timer(2, func).start()      # 实例化一个定时器
    time.sleep(2)               # 等待2S，再重新进入循环

 

★队列【数据安全】-queue

◇先进先出-queue.Queue

方法

• put()　　
• get()
• put_nowait() 满值不阻塞，但报错
• get_nowait() 没值不阻塞，但报错

q = queue.Queue()
q.put('a')
q.put('b')
q.put('c')
print(q.get())
print('ttt:%s' % q.get_nowait())
print('ttt:%s' % q.get_nowait())
print('ttt:%s' % q.get_nowait())    # 队列已空，不阻塞，但报错

◇先进后出-queue.LifoQueue（栈）

方法

• put()
• get()
• put_nowait() 满值不阻塞，但报错
• get_nowait() 没值不阻塞，但报错

q = queue.LifoQueue()
q.put('a')
q.put('b')
q.put('c')
print(q.get())
print('ttt:%s' % q.get_nowait())
print('ttt:%s' % q.get_nowait())
print('ttt:%s' % q.get_nowait())    # 队列已空，不阻塞，但报错

◇优先级队列-queue.PriorityQueue

1， 数字越低，优先级越高；
2， 数字相同，ASCII越小，优先级越高
3， 返回的是一个元祖

方法

• put()
• get()
• put_nowait()  满值不阻塞，但报错
• get_nowait()  没值不阻塞，但报错

from queue import PriorityQueue

q = PriorityQueue()

q.put(30)
q.put(40)
q.put(2)
q.put(2)
q.put(3)
q.put(1)

print(q.get())  # 数字越小，优先级越高
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get_nowait())

★标准模块-concurrent.futures

提供了高度封装的异步调用接口

子模块

ThreadPoolExecutor：线程池，提供异步调用

ProcessPoolExecutor: 进程池，提供异步调用

方法

submit　　异步提交任务

map　　　for循环submit操作

shutdown　　相当于进程池pool.close() + pool.join()

result　　取得结果

add_done_callbak　　回调函数

"""
map： 使用线程池快速实现并发
"""
import time
from concurrent.futures import ThreadPoolExecutor


def fun(n):
    print('in fun %s' % n)
    time.sleep(2)


if __name__ == '__main__':
    thread_pool = ThreadPoolExecutor(max_workers=5)  # 实例化一个线程池，可以同时并发5个线程
    thread_pool.map(fun, range(20))

"""
获取线程池的返回值
"""
import time
from concurrent.futures import ThreadPoolExecutor


def fun(n):
    print('in fun %s' % n)
    time.sleep(2)
    return n * n


if __name__ == '__main__':
    thread_pool = ThreadPoolExecutor(max_workers=5)  # 实例化一个线程池，可以同时并发5个线程
    ret_list = []
    for n in range(20):
        ret = thread_pool.submit(fun, n)
        ret_list.append(ret)
    thread_pool.shutdown()
    print('主线程，打印一下所有线程的返回值', [ret.result() for ret in ret_list])

"""
回调函数的例子
"""
import time
from concurrent.futures import ThreadPoolExecutor


def fun(n):
    print('in fun %s' % n)
    time.sleep(0.3)
    return n * n


def fun_callback(m):
    """
    回调函数
    :param m: 在此示例中为：fun函数的 `return 结果对象`
    :return:
    """
    print('回调函数参数:', m.result())


if __name__ == '__main__':
    thread_pool = ThreadPoolExecutor(max_workers=3)  # 实例化一个线程池，可以同时并发5个线程
    for n in range(20):
        thread_pool.submit(fun, n).add_done_callback(fun_callback)

进程的用法相同，只需要把子模块名字改一下即可