欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

从零开始的Python学习Episode 22——多线程

程序员文章站 2022-07-02 16:42:20
多线程 线程 线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中,是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流,一个进程中可以并发多个线程,每条线程并行执行不同的任务。 进程 程序的执行实例称为进程。 每个进程提供执行程序所需的资源。进程具有虚拟地址空间、可执行代码、系 ......

多线程

 

线程

线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中,是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流,一个进程中可以并发多个线程,每条线程并行执行不同的任务。

 

进程

程序的执行实例称为进程。

每个进程提供执行程序所需的资源。进程具有虚拟地址空间、可执行代码、系统对象的打开句柄、安全上下文、唯一进程标识符、环境变量、优先级类、最小和最大工作集大小以及至少一个执行线程。每个进程都是用一个线程(通常称为主线程)启动的,但是可以从它的任何线程创建额外的线程。

 

线程和进程的区别

线程共享创建它的进程的地址空间;进程有自己的地址空间。

线程可以直接访问其进程的数据段;进程有自己的父进程数据段副本。

线程可以直接与其进程的其他线程通信;进程必须使用进程间通信来与兄弟进程通信。

新线程很容易创建;新进程需要父进程的重复。

线程可以对同一进程的线程进行相当大的控制;进程只能对子进程进行控制。

对主线程的更改(取消、优先级更改等)可能会影响进程的其他线程的行为;对父进程的更改不会影响子进程。

 

线程的创建

import threading
def foo(num):
    print('running on thread ',num)if __name__=='__main__':
    t1 = threading.thread(target=foo, args=(1,))  # 生成一个线程实例
    t1.start()  # 启动线程

 

通过继承的方式创建线程

import threading
class mythread(threading.thread):
    def __init__(self, num):
        threading.thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数,将threading.thread中的run方法进行了重载
        print("running on number:%s" % self.num)

if __name__ == '__main__':
    t1 = mythread(1)
    t2 = mythread(2)
    t1.start()
    t2.start()

 

并发

创建两个线程来同时并发。

import threading
import time
class mythread(threading.thread):
    def __init__(self, num):
        threading.thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数
        print("running on number:%s" % self.num)
        time.sleep(3)

if __name__ == '__main__':
    begin = time.time()
    t1 = mythread(1)
    t2 = mythread(2)
    t1.start()
    t2.start()
    end = time.time()
    print(end-begin)

这样输出的是

running on number:1
running on number:20.002995729446411133

期间只用了零点几秒,几乎是同时输出的,这样可以看出它们是同时进行的。要是是串行的就应该是3.xxxxx秒了。

 

join()方法

在子线程完成运行之前,这个子线程的父线程将一直被阻塞。即一个线程使用join()方法后,必须等该线程结束后才执行join()之后的代码。

import threading
import time
class mythread(threading.thread):
    def __init__(self, num):
        threading.thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数
        print("running on number:%s" % self.num)
        time.sleep(3)

if __name__ == '__main__':
    begin = time.time()
    t1 = mythread(1)
    t2 = mythread(2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    end = time.time()
    print(end-begin)

这样输出的是

running on number:1
running on number:2
3.005915880203247

这样就相当于串行了。

 

守护线程setdaemon

将线程声明为守护线程,必须在start() 方法调用之前设置, 如果不设置为守护线程程序会被无限挂起。这个方法基本和join是相反的。当主线程完成时不需要某个子线程完全运行完就要退出程序,那么就可以将这个子线程设置为守护线程,setdaemon(true)。

import threading
import time
class mythread(threading.thread):
    def __init__(self, num):
        threading.thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数
        print("running on number:%s" % self.num)
        time.sleep(3)

if __name__ == '__main__':
    begin = time.time()
    t1 = mythread(1)
    t2 = mythread(2)
    t1.setdaemon(true)
    t2.setdaemon(true)
    t1.start()
    t2.start()
    end = time.time()
    print(end-begin)

这样当主线程完成了之后就会退出,不会等待子线程。

running on number:1
running on number:20.0020024776458740234

 

同步锁(lock)

import time
import threading

def addnum():
    global num #在每个线程中都获取这个全局变量
    # num-=1

    temp=num
    print('--get num:',num )
    time.sleep(0.001)
    num =temp-1 #对此公共变量进行-1操作


num = 100  #设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.thread(target=addnum)
    t.start()
    thread_list.append(t)

for t in thread_list: #等待所有线程执行完毕
    t.join()

print('final num:', num )

但这样最后获得的final num往往不是0,而是其他数,这是因为多个线程在time.sleep()的时候同时拿到了num,所以num是同一个数,而解决方法就是加锁。

 

死锁与递归锁(rlock)

死锁

import time
import threading
lock1 = threading.lock()
lock2 = threading.lock()
class mythread(threading.thread):
    def run(self):
        self.f1()
        self.f2()
    def f1(self):
        lock1.acquire()
        print('%s 拿到一号锁' %self.name)
        lock2.acquire()
        print('%s 拿到二号锁' % self.name)
        time.sleep(2)
        lock1.release()
        lock2.release()
    def f2(self):
        lock2.acquire()
        print('%s 拿到二号锁' %self.name)
        lock1.acquire()
        print('%s 拿到一号锁' % self.name)
        lock2.release()
        lock1.release()
if __name__ == '__main__':
    for i in range(5):
        t=mythread()
        t.start()

#thread-1 拿到一号锁
thread-1 拿到二号锁
thread-1 拿到二号锁thread-2 拿到一号锁

这里开了5个线程,可是却阻塞住了,原因是在thread1拿到二号锁,thread2拿到一号锁时,f2中在等待一号锁,f1在等待二号锁,结果都等不到,所以产生了死锁。

 

rlock

import time
import threading
lock = threading.rlock()
class mythread(threading.thread):
    def run(self):
        self.f1()
        self.f2()
    def f1(self):
        lock.acquire()
        print('%s 拿到一号锁' %self.name)
        lock.acquire()
        print('%s 拿到二号锁' % self.name)
        time.sleep(2)
        lock.release()
        lock.release()
    def f2(self):
        lock.acquire()
        print('%s 拿到二号锁' %self.name)
        lock.acquire()
        print('%s 拿到一号锁' % self.name)
        lock.release()
        lock.release()
if __name__ == '__main__':
    for i in range(5):
        t=mythread()
        t.start()

'''
thread-1 拿到一号锁
thread-1 拿到二号锁
thread-2 拿到一号锁
thread-2 拿到二号锁
thread-3 拿到一号锁
thread-3 拿到二号锁
thread-4 拿到一号锁
thread-4 拿到二号锁
thread-5 拿到一号锁
thread-5 拿到二号锁
thread-1 拿到二号锁
thread-1 拿到一号锁
thread-2 拿到二号锁
thread-2 拿到一号锁
thread-3 拿到二号锁
thread-3 拿到一号锁
thread-4 拿到二号锁
thread-4 拿到一号锁
thread-5 拿到二号锁
thread-5 拿到一号锁
'''

 

为了支持在同一线程中多次请求同一资源,python提供了“可重入锁”:threading.rlock。rlock内部维护着一个lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release,其他的线程才能获得资源。

 

信号量

信号量用来控制线程并发数的,boundedsemaphore或semaphore管理一个内置的计数 器,每当调用acquire()时-1,调用release()时+1。

      计数器不能小于0,当计数器为 0时,acquire()将阻塞线程至同步锁定状态,直到其他线程调用release()。(类似于停车位的概念)

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

import time
import threading
s = threading.boundedsemaphore(3)
class mythread(threading.thread):
    def run(self):
        s.acquire()
        print(self.name)
        time.sleep(2)
        s.release()
if __name__ == '__main__':
    t=[]
    for i in range(10):
        t.append(mythread())
    for i in t:
        i.start()

'''
thread-1
thread-2
thread-3
thread-4thread-6thread-5


thread-7
thread-8thread-9

thread-10
'''

可以看到几乎是三个三个同时输出的。

 

条件变量同步

有一类线程需要满足条件之后才能够继续执行,python提供了threading.condition 对象用于条件变量线程的支持,它除了能提供rlock()或lock()的方法外,还提供了 wait()、notify()、notifyall()方法。

      lock_con=threading.condition([lock/rlock]): 锁是可选选项,不传入锁,对象自动创建一个rlock()。

wait():条件不满足时调用,线程会释放锁并进入等待阻塞;
notify():条件创造后调用,通知等待池激活一个线程;
notifyall():条件创造后调用,通知等待池激活所有线程。
import time
import threading
import random
lock = threading.condition()
class producer(threading.thread):
    def __init__(self,name):
        threading.thread.__init__(self)
        self.name = name
    def run(self):
        global product
        while true:
            if lock.acquire():
                p = random.randint(1,10)
                print('机器%s生产了%d件产品'%(self.name,p))
                product+=p
                lock.notify()
                lock.release()
                time.sleep(2)

class consumer(threading.thread):
    def __init__(self,name):
        threading.thread.__init__(self)
        self.name = name
    def run(self):
        global product
        while true:
            if lock.acquire():
                if product>=1:
                    print('客户%s购买了1件产品'%self.name)
                    product -=1
                    lock.notify()
                    lock.release()
            time.sleep(2)


if __name__=="__main__":
    product = 0
    t = []
    for i in range(5):
        t.append(producer(i))
    t.append((consumer(1)))
    t.append(consumer(2))
    for x in t:
        x.start()

'''
机器0生产了10件产品
机器1生产了4件产品
机器2生产了1件产品
机器3生产了2件产品
机器4生产了3件产品
客户1购买了1件产品
客户2购买了1件产品
机器0生产了8件产品
机器1生产了7件产品
机器2生产了5件产品
机器3生产了1件产品
机器4生产了4件产品
客户2购买了1件产品
客户1购买了1件产品
机器0生产了8件产品
机器1生产了4件产品
机器2生产了5件产品
机器3生产了1件产品
机器4生产了1件产品
客户1购买了1件产品
客户2购买了1件产品
机器0生产了5件产品
机器1生产了2件产品
'''

wait等待notify的通知,当接到通知后,会重新从if acquire()开始执行

 

同步条件(event)

 

 

event.isset():返回event的状态值;

event.wait():如果 event.isset()==false将阻塞线程;

event.set(): 设置event的状态值为true,所有阻塞池的线程激活进入就绪状态,等待操作系统调度;

event.clear():恢复event的状态值为false。
import time
import threading
import random
event1=threading.event()
event2=threading.event()
event3=threading.event()
class producer(threading.thread):
    def run(self):
        event1.wait()
        print('salesman:这个东西100块。')
        print('salesman:你要吗?')
        event1.clear()
        event2.set()
        event3.wait()
        print('salesman:好的')
        event3.clear()
class consumer(threading.thread):
    def run(self):
        print('customer:这个东西多少钱?')
        event1.set()
        event2.wait()
        print('customer:嗯,帮我包起来')
        event2.clear()
        event3.set()




if __name__=="__main__":
    t = []
    t.append(producer())
    t.append(consumer())
    for x in t:
        x.start()

'''
customer:这个东西多少钱?
salesman:这个东西100块。
salesman:你要吗?
customer:嗯,帮我包起来
salesman:好的
'''

 

队列queue

创建一个“队列”对象
import queue
q = queue.queue(maxsize = 10)
queue.queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。


将一个值放入队列中
q.put(10)
调用队列对象的put()方法在队尾插入一个项目。put()有两个参数,第一个item为必需的,为插入项目的值;第二个block为可选参数,默认为
1。如果队列当前为空且block为1,put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0,put方法将引发full异常。


将一个值从队列中取出
q.get()
调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block,默认为true。如果队列为空且block为true,get()就使调用线程暂停,直至有项目可用。如果队列为空且block为false,队列将引发empty异常。


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,queue
from time import sleep
from random import randint
class production(threading.thread):
    def run(self):
        while true:
            r=randint(0,100)
            q.put(r)
            print("生产出来%s号包子"%r)
            sleep(1)
class proces(threading.thread):
    def run(self):
        while true:
            re=q.get()
            print("吃掉%s号包子"%re)
if __name__=="__main__":
    q=queue.queue(10)
    threads=[production(),production(),production(),proces()]
    for t in threads:
        t.start()

'''
生产出来55号包子
生产出来100号包子
生产出来67号包子
吃掉55号包子
吃掉100号包子
吃掉67号包子
生产出来23号包子生产出来97号包子生产出来41号包子吃掉23号包子



吃掉97号包子
吃掉41号包子
吃掉4号包子
生产出来4号包子
生产出来45号包子吃掉45号包子生产出来84号包子
'''