Python 多线程多进程

Posted 2022-02-05 py-peng

什么是线程？

py文件在执行程序中，他会根据程序的编写来区分，假如没有创建子进程，整个程序就是主进程。

那程序中，有主线程而且还有子线程，那他就是一个多线程。

使用多线程可以提升I/O密集型的效率。

什么是进程？

py文件就是一个进程，比如：QQ，360，浏览器。

使用多进程，会消耗很大的资源问题。

GIL锁

GIL锁又称，全局解释器锁。

GIL锁的作用：在同一时刻，只能有一个线程进入解释器。

站在开发Python语言的那一端时，他就是一个神器，而站在使用这门语言这一端时，他就是一个BUG般的存在，而这BUG仅存在于CPython中。

为什么说BUG呢，因为有了GIL锁，我们使用多线程在进行计算密集型中，计算机的核数再多，他也只能使用一核。

I/O密集型，计算密集型

什么是I/O密集型？

说白了，他就是一个堵塞，当我们创建多线程（A、B），A线程在执行，遇到了堵塞，在CPU空闲时，切换到了B线程。

import threading
import time

start  = time.time()
def music():
    print(‘I listening Music‘)
    time.sleep(2)

def movie():
    print(‘I watching TV‘)
    time.sleep(3)

t1 = threading.Thread(target=music)
t2 = threading.Thread(target=movie)
t1.start()
t2.start()
end = time.time()
result = end - start
print(result)
print(‘主线程结束‘)

View Code

什么时计算密集型？

线程在计算过程中，没有遇到堵塞，而是一直在执行计算。

def add():
    num = 0
    for i in range(1000000):
        num += i
    print(num)

View Code

如何创建多线程？

 Threading模块

函数创建

import threading
import time

start  = time.time()
def music():
    for i in range(3):
        print(‘I listening Music‘)
        time.sleep(1)

def movie():
    for i in range(2):
        print(‘I watching TV‘)
        time.sleep(5)

t1 = threading.Thread(target=music) #创建子线程
t2 = threading.Thread(target=movie) #创建子线程
threads = [t1,t2]

for t in threads:
    t.start() #启动子线程

View Code

类创建

import threading

class MyThread(threading.Thread): # 首先要继承这个方法
    def __init__(self,count):
        super().__init__()
        self.count = count
    def current_thread_count(self):
        print(self.count)
    def run(self): #定义每个线程要运行的内容
        self.current_thread_count()

t1 = MyThread(threading.active_count())
t2 = MyThread(threading.active_count())
t1.start() #开启线程
t2.start()

View Code

join ()方法

主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，

才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。join([timeout]) 里面的参数时可选的，代表线程运行的最大时

间，即如果超过这个时间，不管这个此线程有没有执行完毕都会被回收，然后主线程或函数都会接着执行的，如果线程执行时间小于参数表示的

时间，则接着执行，不用一定要等待到参数表示的时间。

import threading
import time

start  = time.time()
def music():
    for i in range(3):
        print(‘I listening Music‘)
        time.sleep(1)

def movie():
    for i in range(2):
        print(‘I watching TV‘)
        time.sleep(5)

t1 = threading.Thread(target=music) #创建子线程
t2 = threading.Thread(target=movie) #创建子线程
threads = [t1,t2]

for t in threads:
    t.start() #启动子线程
t.join() # 代表赋值前的一个，也就是t2
print(‘主线程结束‘)

View Code

setDaemon()方法

主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这

时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。此外，还有

个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起，只有等待了所有线程结束它才结束。

import threading
import time

start  = time.time()
def music():
    for i in range(3):
        print(‘I listening Music‘)
        time.sleep(1)

def movie():
    for i in range(2):
        print(‘I watching TV‘)
        time.sleep(5)

t1 = threading.Thread(target=music) #创建子线程
t2 = threading.Thread(target=movie) #创建子线程
threads = [t1,t2]
t2.setDaemon(1) #守护线程

for t in threads:
    t.start() #启动子线程
print(‘主线程结束‘)

View Code

同步锁

为什么会有同步锁？

当我们创建多线程时，并且有一个全局变量，而多线程操作这个全局变量。

import threading
import time

def sub():
    global number
    num = number
    time.sleep(0.1)
    number = num - 1

number = 10
threads = []

for i in range(10):
    t = threading.Thread(target=sub)
    t.start()
    threads.append(t)
for i in threads:
    i.join()
print(number) # 9

View Code

结果并不是我们想要的。

为什么出现这种问题？

程序在sleep的一瞬间，cpu来回切换，还没等着修改全局变量，所有的线程已经被创建，而且也已经被赋值。

如何解决？

那就是加锁了。

import threading
import time

def sub():
    global number

    r.acquire() # 获得锁
    num = number
    time.sleep(0.1)
    number = num - 1
    r.release() # 释放锁

number = 10
threads = []
r = threading.Lock()

for i in range(10):
    t = threading.Thread(target=sub)
    t.start()
    threads.append(t)
for i in threads:
    i.join()
print(number) # 0

View Code

加锁，其实就是不让cup进行线程切换，直到锁被释放。

如果锁没被释放，不会让其他线程进入，也不会影响不进入线程的执行。

import threading
import time

number = 10
threads = []
r = threading.Lock()

def sub():
    global number
    r.acquire()
    num = number
    time.sleep(0.1)
    number = num - 1
    r.release()

def music():
    time.sleep(0.5)
    print(‘Music‘)

t = threading.Thread(target=music)
t.start()
for i in range(10):
    t = threading.Thread(target=sub)
    t.start()
    threads.append(t)
for i in threads:
    i.join()
print(number)

View Code

递归锁(Rlock)

import threading

r = threading.Lock()

class MyThread(threading.Thread):
    def Thread_1(self):
        r.acquire()
        print(‘第一层‘,self.name)
        r.acquire()
        print(‘第二层‘,self.name)
        r.release()
        r.release()
    def run(self):
        self.Thread_1()

for i in range(5):
    t = MyThread()
    t.start()

死锁

递归锁，与Lock很相似，但是他有一个计数的功能，能解决死锁

import threading

r = threading.RLock()

class MyThread(threading.Thread):
    def Thread_1(self):
        r.acquire()
        print(‘第一层‘,self.name)
        r.acquire()
        print(‘第二层‘,self.name)
        r.release()
        r.release()
    def Thread_2(self):
        r.acquire()
        print(‘第一层‘,self.name)
        r.acquire()
        print(‘第二层‘,self.name)
        r.release()
        r.release()
    def run(self):
        self.Thread_1()
        self.Thread_2()

for i in range(5):
    t = MyThread()
    t.start()

View Code

信号量(Semaphore)

信号量相当于，可以限制最大进入的线程数量。

import threading
import time
r = threading.Semaphore(2) # 创建信号量，最大进入的线程数量

class MyThread(threading.Thread):
    def Thread_1(self):

        r.acquire() # 每次进入线程+1，但不能超过信号量设定的值
        print(self.name)
        time.sleep(2)
        r.release() # -1
    def run(self):
        self.Thread_1()

for i in range(5):
    t = MyThread()
    t.start()

View Code

条件变量(Conditon) 

wait()：条件不满足时调用，线程会释放锁并进入等待阻塞。

notify()：条件创造后调用，通知等待池激活一个线程。

notifyAll()：条件创造后调用，通知等待池激活所有线程。

import threading
import time
import random

def producer():
    time.sleep(0.2)
    global F
    while True:
        if con_Lock.acquire():
            r = random.randrange(0,100)
            F.append(r)
            print(str(threading.current_thread())+ ‘--->‘ + str(r))
            con_Lock.notify()
            con_Lock.release()
        time.sleep(3)

def consumer():
    global F
    while True:
        con_Lock.acquire()
        if not F:
            print("老板，快点，没有包子了")
            con_Lock.wait()
        a = F.pop()
        print(‘包子%s已经被吃‘%a)
        time.sleep(0.5)

con_Lock = threading.Condition()
threads = []
F = []

for i in range(5):
    threads.append(producer)
threads.append(consumer)
for i in threads:
    t = threading.Thread(target=i)
    t.start()

View Code

队列(Queue)

Python Queue模块有三种队列及构造函数:
1、Python Queue模块的FIFO队列先进先出。  class queue.Queue(maxsize)
2、LIFO类似于堆，即先进后出。             class queue.LifoQueue(maxsize)
3、还有一种是优先级队列级别越低越先出来。   class queue.PriorityQueue(maxsize)

此包中的常用方法(q = Queue.Queue()):
q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)
非阻塞 q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

import threading,time

li=[1,2,3,4,5]

def pri():
    while li:
        a=li[-1]
        print(a)
        time.sleep(1)
        try:
            li.remove(a)
        except:
            print(‘----‘,a)

t1=threading.Thread(target=pri,args=())
t1.start()
t2=threading.Thread(target=pri,args=())
t2.start()

View Code

import queue
import threading
import time
import random

q = queue.Queue()

class Do(threading.Thread):
    def __init__(self):
        super().__init__()
        self.mean = [‘白菜‘,‘糯米‘,‘大葱‘,‘红豆‘,‘绿豆‘,‘羊肉‘,‘猪肉‘,‘鸡蛋‘,‘青椒‘]
    def run(self):
        time.sleep(0.2)
        while True:
            ran = random.choice(self.mean)
            q.put(ran)
            time.sleep(0.1)

class Eat(threading.Thread):
    def run(self):
        while True:
            if not q.empty():
                print(‘这个包子是：%s‘%q.get())
            else:
                print(‘老板，没有包子了‘)
            time.sleep(0.5)

threads = []
for i in range(5):
    threads.append(Eat())
    threads.append(Do())
for t in threads:
    t.start()

View Code