本站消息

  本站每日ip已达10000,出租广告位,位置价格可谈,需要合作请联系站长


+关注
已关注

分类  

暂无分类

标签  

暂无标签

日期归档  

2019-07(5)

2019-08(103)

2019-09(106)

2019-10(16)

2019-11(6)

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

发布于2019-08-08 16:21     阅读(200)     评论(0)     点赞(2)     收藏(3)


多线程

 

线程

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

 

进程

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

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

 

线程和进程的区别

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

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

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

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

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

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

 

线程的创建

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号包子
'''
复制代码

 



所属网站分类: 技术文章 > 博客

作者:喜洋洋与红太狼

链接: https://www.pythonheidong.com/blog/article/15203/

来源: python黑洞网

任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任

2 0
收藏该文
已收藏

评论内容:(最多支持255个字符)