Python中的线程和进程

标签：mon   运行   RoCE   完成后   input   进程和线程   space   cti   owa   

引入进程和线程的概念及区别

threading模块提供的类： 　　Thread, Lock, Rlock, Condition, [Bounded]Semaphore, Event, Timer, local。

1.什么是进程

计算机程序只不过是磁盘中可执行的二进制（或其他类型）的数据。它们只有在被读取到内存中，被操作系统调用的时候才开始它们的生命期。

进程（有时被称为重量级进程）是程序的一次执行。每个进程都有自己的地址空间、内存、数据栈及其它记录其运行轨迹的辅助数据。

操作系统管理在其上运行的所有进程，并为这些进程公平的分配时间，进程也可以通过fork和spawn操作来完成其它的任务。

不过各个进程有自己的内存空间、数据栈等，所以只能使用进程间通讯，而不能直接共享信息。

2.线程的基本概念

线程是进程中执行运算的最小单位，是进程中的一个实体，是被系统独立调度和分派的基本单位，线程自己不拥有系统资源，只拥有一点在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程，同一进程中的多个线程之间可以并发执行。

3、线程和进程的关系以及区别？

进程和线程的关系：

• （1）一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程。
• （2）资源分配给进程，同一进程的所有线程共享该进程的所有资源。
• （3）处理机分给线程，即真正在处理机上运行的是线程
• （4）线程在执行过程中，需要协作同步。不同进程的线程间要利用消息通信的办法实现同步。线程是指进程内的一个执行单元,也是进程内的可调度实体.

进程与线程的区别：

• （1）调度：线程作为调度和分配的基本单位，进程作为拥有资源的基本单位
• （2）并发性：不仅进程之间可以并发执行，同一个进程的多个线程之间也可并发执行
• （3）拥有资源：进程是拥有资源的一个独立单位，线程不拥有系统资源，但可以访问隶属于进程的资源.
• （4）系统开销：在创建或撤消进程时，由于系统都要为之分配和回收资源，导致系统的开销明显大于创建或撤消线程时的开销。

1 多进程创建方式

可以归纳为三种：fork，multiprocessing以及进程池Pool。

（1） fork方式

import os

# 注意，fork函数，只在Unix/Linux/Mac上运行，windows不可以
pid = os.fork()
print(pid)
if pid == 0:
    print(‘哈哈1‘)
else:
    print(‘哈哈2‘)
    
#1880
#哈哈2
#0
#哈哈1

注意：fork()函数只能在Unix/Linux/Mac上面运行，不可以在Windows上面运行。

说明：

• 程序执行到os.fork()时，操作系统会创建一个新的进程（子进程），然后复制父进程的所有信息到子进程中
• 然后父进程和子进程都会从fork()函数中得到一个返回值，在子进程中这个值一定是0，而父进程中是子进程的 id号

在Unix/Linux操作系统中，提供了一个fork()系统函数，它非常特殊。

普通的函数调用，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。

子进程永远返回0，而父进程返回子进程的ID。

这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。我们可以通过os.getpid()获取当前进程ID，通过os.getppid()获取父进程ID。

那么，父子进程之间的执行有顺序吗？

答案是没有！这完全取决于操作系统的调度算法。

而多次fork()就会产生一个树的结构：

（2）multiprocessing方式

如果你打算编写多进程的服务程序，Unix/Linux无疑是正确的选择。由于Windows没有fork调用，难道在Windows上无法用Python编写多进程的程序？当然可以！由于Python是跨平台的，自然也应该提供一个跨平台的多进程支持。multiprocessing模块就是跨平台版本的多进程模块。

import  os
import time

from multiprocessing import  Process

def run_proc(name):
    print(‘子进程运行中，name=%s,pin=%d...‘%(name,os.getpid()))

    time.sleep(10)
    print(‘子进程已经结束‘)

if __name__==‘__main__‘:
    print(‘父进程%d.‘%os.getpid())
    p=Process(target=run_proc,args=(‘test‘,))
    print(‘子进程将要执行‘)
    p.start()


# 父进程19196.
# 子进程将要执行
# 子进程运行中，name=test,pin=8160...
# 子进程已经结束

从结果我们看出，只要通过start()开启了子进程之后，主进程会等待子进程执行完才结束！

Process的语法结构如下：

import  os
import time
from multiprocessing import Process


class MyProcess(Process):
    def __init__(self,name):
        Process.__init__(self)
        self.name=name

    def run(self):
        print(‘子进程运行中，name=%s ,pid=%d...‘ % (self.name, os.getpid()))
        import time
        time.sleep(10)
        print(‘子进程已结束‘)
        

if __name__=="__main__":
    my=MyProcess(‘test‘)
    my.start()


# 子进程运行中，name=test ,pid=11676...
# 子进程已结束

（3）Pool方式

#coding=utf-8
from multiprocessing import Pool
import os, time, random

def worker(msg):
    print("%s开始执行,进程号为%d"%(msg, os.getpid()))
    time.sleep(1)
    print("%s执行完毕"%(msg))

if __name__ == ‘__main__‘:
    po = Pool(3)  # 定义一个进程池，最大进程数3
    for i in range(10):
        # Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))
        # 每次循环将会用空闲出来的子进程去调用目标
        po.apply_async(worker, (i,))

    print("----start----")
    po.close()  # 关闭进程池，关闭后po不再接收新的请求
    po.join()  # 等待po中所有子进程执行完成，必须放在close语句之后
    print("-----end-----")


# ----start----
# 1开始执行,进程号为21192
# 1执行完毕
# 4开始执行,进程号为21192
# 4执行完毕
# 7开始执行,进程号为21192
# 7执行完毕
# 2开始执行,进程号为10596
# 2执行完毕
# 5开始执行,进程号为10596
# 5执行完毕
# 8开始执行,进程号为10596
# 8执行完毕
# 0开始执行,进程号为17116
# 0执行完毕
# 3开始执行,进程号为17116
# 3执行完毕
# 6开始执行,进程号为17116
# 6执行完毕
# 9开始执行,进程号为17116
# 9执行完毕
# -----end-----

实现一个多进程下的文件夹复制功能：

#coding=utf-8

import os
from multiprocessing import Pool


def copyFileTask(name, oldFolderName, newFolderName):
    # 完成copy一个文件的功能
    fr = open(oldFolderName+"/"+name, ‘rb+‘)
    fw = open(newFolderName+"/"+name, ‘wb+‘)

    str = fr.read(1024 * 5)
    while (str != ‘‘):
        fw.write(str)
        str = fr.read(1024 * 5)

    fr.close()
    fw.close()


def main():
    # 获取要copy的文件夹名字
    oldFolderName = raw_input(‘请输入文件夹名字：‘)
    # 创建一个文件夹
    newFolderName = oldFolderName+‘-复件‘.decode(‘utf-8‘).encode(‘gbk‘)
    os.mkdir(newFolderName)

    #获取old文件夹里面所有文件的名字
    fileNames = os.listdir(oldFolderName)

    #使用多进程的方式copy原文件夹所有内容到新的文件夹中
    pool = Pool(5)
    for name in fileNames:
        pool.apply_async(copyFileTask, (name, oldFolderName, newFolderName))

    pool.close()
    pool.join()

if __name__ == ‘__main__‘:
    main()

(4)python进程间通信-Queue

from multiprocessing import Queue,Process
import  time

def write(q):
    for i in ["A","B","C","D","E"]:
        print(‘向队列中添加%s‘%i)
        q.put(i)
        #print(id(q))
        time.sleep(1)

def read(q):
    #print(q.empty())
    while not q.empty():
        print("从队列中取出来的值是%s"%q.get())
        time.sleep(1)


if __name__ == ‘__main__‘:
    q=Queue()#创建一个queue队列，当做参数输入子进程中，这样子进程能拿到数据资源
    qw=Process(target=write,args=(q,))
    qw.start()
    qw.join(0.1)

    qr=Process(target=read,args=(q,))
    qr.start()
    qr.join()
    print("通信完毕")

# Queue-1
# 
# 向队列中添加A
# 向队列中添加B
# 向队列中添加C
# 向队列中添加D
# 向队列中添加E
# 从队列中取出来的值是A
# 从队列中取出来的值是B
# 从队列中取出来的值是C
# 从队列中取出来的值是D
# 从队列中取出来的值是E
# 通信完毕

(1) 在实例化Queue类，可以传递最大消息数，如q = Queue(5)，这段代码是指只允许消息队列中最大有5个消息数据。如果不加最大消息数或数量为负值，则表达不限制数量直到占满内存；
(2) Queue.qsize()：返回当前队列包含的消息数量；
(3) Queue.empty()：如果队列为空，返回True，反之False ；
(4) Queue.full()：如果队列满了，返回True,反之False；
(5) Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True；
1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出”Queue.Empty”异常；
2）如果block值为False，消息列队如果为空，则会立刻抛出”Queue.Empty”异常；
(6) Queue.get_nowait()：相当Queue.get(False)；
(7) Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为True；
1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息列队腾出空间为止，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出”Queue.Full”异常；
2）如果block值为False，消息列队如果没有空间可写入，则会立刻抛出”Queue.Full”异常；
(8) Queue.put_nowait(item)：相当Queue.put(item, False)；

二：进程池(pool)中的Queue

如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否则会得到一条如下的错误信息：
RuntimeError: Queue objects should only be shared between processes through inheritance.

from multiprocessing import Manager,Pool
import time

def write(q):
    for i in ["A","B","C","D","E"]:
        print("向队列中添加%s"%i)
        q.put(i)
        time.sleep(1)

def read(q):
    while not q.empty():
        print("从队列中取出的值是%s"%q.get())
        time.sleep(1)

if __name__ == ‘__main__‘:
    q = Manager().Queue()
    pool = Pool()
    pool.apply_async(write,args=(q,))
    pool.apply_async(read,args=(q,))
    # pool.apply(write,args=(q,))
    # pool.apply(read,args=(q,))

    pool.close()
    pool.join()

    print("数据通信完毕")

# Queue-pool
# 
# 向队列中添加A
# 向队列中添加B
# 向队列中添加C
# 向队列中添加D
# 向队列中添加E
# 从队列中取出的值是A
# 从队列中取出的值是B
# 从队列中取出的值是C
# 从队列中取出的值是D
# 从队列中取出的值是E
# 数据通信完毕

(5)管道pipe

pipe()返回两个连接对象代表pipe的两端。每个连接对象都有send()方法和recv()方法。

但是如果两个进程或线程对象同时读取或写入管道两端的数据时，管道中的数据有可能会损坏。

当进程使用的是管道两端的不同的数据则不会有数据损坏的风险。

使用方法：

from multiprocessing import  Process,Pipe
def f(conn):
    conn.send(‘约吗‘)
    conn.send(‘约吗.....‘)
    print(conn.recv())#接收数据
    conn.close()

if __name__=="__main__" :
    parent_conn,chid_conn=Pipe()
    p=Process(target=f,args=(chid_conn,))
    p.start()
    print(parent_conn.recv())#接收数据
    print(parent_conn.recv())
    parent_conn.send(‘约啊‘)#发送数据

# 约吗
# 约吗.....
# 约啊

(6)Manager

　进程间的通信Queue()和Pipe(),可以实现进程间的数据传递。但是要使python进程间共享数据，我们就要使用multiprocessing.Manager。

　　Manager()返回的manager对象控制了一个server进程，此进程包含的python对象可以被其他的进程通过proxies来访问。从而达到多进程间数据通信且安全。

　　Manager支持list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。

from multiprocessing import Process, Manager
import os
 
 
def f(d, l):
    d[os.getpid()] = os.getpid()   # 获取进程号
    l.append(os.getpid())
    print(l)
 
 
if __name__ == ‘__main__‘:
    with Manager() as manager:
        d = manager.dict()   # 生成一个字典，可以在多个进程间共享和传递
        l = manager.list(range(5))  # 生成一个列表，可以在多个进程间共享和传递
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
 
        print(d)
        print(l)

# [0, 1, 2, 3, 4, 21300]
# [0, 1, 2, 3, 4, 21300, 4828]
# [0, 1, 2, 3, 4, 21300, 4828, 19088]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568, 21636]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568, 21636, 21664]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568, 21636, 21664, 21684]
# [0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568, 21636, 21664, 21684, 21736]
# {21300: 21300, 4828: 4828, 19088: 19088, 18628: 18628, 21520: 21520, 21568: 21568, 21636: 21636, 21664: 21664, 21684: 21684, 21736: 21736}
[0, 1, 2, 3, 4, 21300, 4828, 19088, 18628, 21520, 21568, 21636, 21664, 21684, 21736]

进程间的数据默认不共享，那Manager()如何来同步数据？manager对象控制了一个server进程，这个对象有共享数据和让其他进程能访问数据的代理。一个进程修改了数据后，其实并不会通过manager传送数据，作为代理而言，它不知道数据被改变了。为了在共享数据中同步修改的内容，需要在这些代理的容器中，重新声明这个修改过的内容。

2. Python中的线程

Python的标准库提供了两个模块：thread和threading，thread是低级模块，threading是高级模块，对thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。

from time import  ctime,sleep
import threading

def music(func):
    print(threading.current_thread())
    for i in range(2):
        print(‘I was listening to %s. %s‘%(func,ctime()))
        sleep(2)
        print(‘end listing %s‘%ctime())

t1=threading.Thread(target=music,args=(‘星晴‘,))#创建线程，traget执行的函数，args函数参数
t1.start()#开启线程

更多方法：

• start 线程准备就绪，等待CPU调度
• setName 为线程设置名称
• getName 获取线程名称
• setDaemon 设置为后台线程或前台线程（默认）;如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否,均停止;如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
• join 逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
• run 线程被cpu调度后自动执行线程对象的run方法
• Lock 线程锁(互斥锁Mutex)
• Event

线程交互执行,Join(),Daemon ：

1、join ()方法：主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。

2、setDaemon()方法。主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。此外，还有个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。

from time import  ctime,sleep
import threading

def music(func):
    print(threading.current_thread())
    for i in range(2):
        print(‘I was listening to %s. %s‘%(func,ctime()))
        sleep(2)
        print(‘end listing %s‘%ctime())
def move(func):
    print(threading.current_thread())
    for i in range(2):
        print(‘I was at the %s ! %s‘%(func,ctime()))
        sleep(3)
        print(‘end moving %s‘ % ctime())

threads=[]
t1=threading.Thread(target=music,args=(‘星晴‘,))
threads.append(t1)
t2=threading.Thread(target=move,args=(‘正义联盟‘,))
threads.append(t2)

#join（）
if __name__=="__main__":
    # music(u‘七里香‘)
    # move(u‘功夫‘)
    for t in threads:
        t.start()
    t.join()#等价于t2.join，目的是为了最后执行最后一条打印信息，join 逐个执行每个线程，执行完毕后继续往下执行，
    #只有当所有线程执行完毕后才会执行下一个
    print(‘all over %s‘%ctime())

from time import  ctime,sleep
import threading

def music(func):
    #print(threading.current_thread())#线程对象
    for i in range(2):
        print(‘I was listening to %s. %s‘%(func,ctime()))
        sleep(2)
        print(‘end listing %s‘%ctime())
def move(func):
    #print(threading.current_thread())#线程对象
    for i in range(2):
        print(‘I was at the %s ! %s‘%(func,ctime()))
        sleep(3)
        print(‘end moving %s‘ % ctime())

threads=[]
t1=threading.Thread(target=music,args=(‘星晴‘,))
threads.append(t1)
t2=threading.Thread(target=move,args=(‘正义联盟‘,))
threads.append(t2)


#Daemon(守护进程)将主线程设置为Daemon线程,它退出时,其它子线程会同时退出,不管是否执行完任务。
if __name__=="__main__":
    t2.setDaemon(True)#只守护t2，但不守护t1，t1会全部执行完，同时也会引发t2执行，但t2不会执行完，t1执行完，就退出
    for t in threads:
        #t.setDaemon(True)#（执行一次后就退出）
        t.start()


    #print(threading.current_thread())#线程对象
    #print(threading.active_count())#线程对象的数量（默认会有一条主线程）
    print(‘all over %s‘%ctime())

线程继承实例化：

import threading
import time
#t1=threading.Thread(target=music,args=(‘星晴‘,))
#继承式调用
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__‘:
    t1=MyThread(1)
    t2=MyThread(2)
    t1.start()
    t2.start()

3.线程锁(互斥锁Mutex)

1.创建锁:mutex = threading.Lock() 2.锁定:mutex.acquire([timeout]) 3.释放:mutex.release()

import time
import threading

def addNum():
    global num #在每个线程中都获取这个全局变量
    print(‘--get num:‘,num )
    time.sleep(1)
    num  -=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 )

正常来讲，这个num结果应该是0， 但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？ 哈，很简单，假设你有A,B两个线程，此时都 要对num 进行减1操作， 由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。那怎么办呢？ 很简单，每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。

注：不要在3.x上运行，不知为什么，3.x上的结果总是正确的，可能是自动加了锁

加锁版本：

import time
import threading

def addNum():
    global num #在每个线程中都获取这个全局变量
    print(‘--get num:‘,num )
    time.sleep(1)
    lock.acquire() #修改数据前加锁
    num  -=1 #对此公共变量进行-1操作
    lock.release() #修改后释放

num = 100  #设定一个共享变量
thread_list = []
lock = threading.Lock() #生成全局锁
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 )

4.死锁和递归锁

所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，

它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，

如下就是死锁:

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print(‘\033[41m%s 拿到A锁\033[0m‘ %self.name)

        mutexB.acquire()
        print(‘\033[42m%s 拿到B锁\033[0m‘ %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print(‘\033[43m%s 拿到B锁\033[0m‘ %self.name)
        time.sleep(2)

        mutexA.acquire()
        print(‘\033[44m%s 拿到A锁\033[0m‘ %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == ‘__main__‘:
    for i in range(5):
        t=MyThread()
        t.start()

输出结果如下：

Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁

分析如上代码是如何产生死锁的：
启动5个线程，执行run方法，假如thread1首先抢到了A锁，此时thread1没有释放A锁，紧接着执行代码mutexB.acquire()，抢到了B锁，在抢B锁时候，没有其他线程与thread1争抢，因为A锁没有释放，其他线程只能等待，然后A锁就执行完func1代码，然后继续执行func2代码，与之同时，在func2中，执行代码 mutexB.acquire()，抢到了B锁，然后进入睡眠状态，在thread1执行完func1函数，释放AB锁时候，其他剩余的线程也开始抢A锁，执行func1代码，如果thread2抢到了A锁，接下来thread2要抢B锁，ok，在这个时间段，thread1已经执行func2抢到了B锁，然后在sleep(2),持有B锁没有释放，为什么没有释放，因为没有其他的线程与之争抢，他只能睡着，然后thread1握着B锁，thread2要抢B锁，ok，这样就形成了死锁

递归锁

我们分析了死锁，那么python里面是如何解决这样的递归锁呢？
在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

from threading import Thread,Lock,RLock
import time

mutexA=mutexB=RLock()


class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print(‘%s 拿到A锁‘ %self.name)

        mutexB.acquire()
        print(‘%s 拿到B锁‘ %self.name)
        mutexB.release()

        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print(‘%s 拿到B锁‘ % self.name)
        time.sleep(0.1)
        mutexA.acquire()
        print(‘%s 拿到A锁‘ % self.name)
        mutexA.release()

        mutexB.release()


if __name__ == ‘__main__‘:
    for i in range(5):
        t=MyThread()
        t.start()

输出结果：

Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-1 拿到A锁
Thread-2 拿到A锁
Thread-2 拿到B锁
Thread-2 拿到B锁
Thread-2 拿到A锁
Thread-4 拿到A锁
Thread-4 拿到B锁
Thread-4 拿到B锁
Thread-4 拿到A锁
Thread-3 拿到A锁
Thread-3 拿到B锁
Thread-3 拿到B锁
Thread-3 拿到A锁
Thread-5 拿到A锁
Thread-5 拿到B锁
Thread-5 拿到B锁
Thread-5 拿到A锁

来解释下递归锁的代码：
由于锁A，B是同一个递归锁，thread1拿到A,B锁，counter记录了acquire的次数2次，然后在func1执行完毕，就释放递归锁，在thread1释放完递归锁，执行完func1代码，接下来会有2种可能，1、thread1在次抢到递归锁，执行func2代码 2、其他的线程抢到递归锁，去执行func1的任务代码

5.信号量Semaphore（其实也是一把锁）

Semaphore管理一个内置的计数器

Semaphore与进程池看起来类似，但是是完全不同的概念。

进程池：Pool(4)，最大只能产生四个进程，而且从头到尾都只是这四个进程，不会产生新的。

信号量：信号量是产生的一堆进程/线程，即产生了多个任务都去抢那一把锁

 1 from threading import Thread,Semaphore,currentThread
 2 import threading
 3 import time,random
 4 sm=Semaphore(5)#运行的时候有5个人
 5 def task():
 6     sm.acquire()
 7     print(‘\033[42m %s上厕所‘ % currentThread().getName())
 8     time.sleep(random.randint(1,3))
 9     print(‘\033[31m %s上完厕所走了‘ % currentThread().getName())
10     sm.release()
11 
12 if __name__=="__main__":
13     for i in range(20):#开了10个线程 ，这20人都要上厕所
14         t=Thread(target=task)
15         t.start()
16 
17 
18 /*********************
19 import threading
20 from threading import Thread,Semaphore,currentThread
21 #继承实例化
22 class MyThread(threading.Thread):
23     def run(self):
24         if semaphore.acquire():
25             print(self.name)
26             time.sleep(3)
27             semaphore.release()
28 
29 if __name__=="__main__":
30     semaphore=threading.Semaphore(5)
31     thrs=[]
32     for i in range(23):
33         thrs.append(MyThread())
34     for t in thrs:
35         t.start()

输出结果：

D:\ProgramData\Anaconda3\python.exe C:/Users/Administrator/PycharmProjects/untitled1/week4/进程与线程/锁的使用.py
 Thread-1上厕所
 Thread-2上厕所
 Thread-3上厕所
 Thread-4上厕所
 Thread-5上厕所
 Thread-1上完厕所走了
 Thread-4上完厕所走了
 Thread-7上厕所
 Thread-6上厕所
 Thread-5上完厕所走了
 Thread-8上厕所
 Thread-3上完厕所走了
 Thread-2上完厕所走了
 Thread-6上完厕所走了
 Thread-9上厕所
 Thread-10上厕所
 Thread-11上厕所
 Thread-8上完厕所走了
 Thread-12上厕所
 Thread-9上完厕所走了
 Thread-7上完厕所走了
 Thread-14上厕所
 Thread-13上厕所
 Thread-14上完厕所走了
 Thread-13上完厕所走了
 Thread-16上厕所
 Thread-15上厕所
 Thread-12上完厕所走了
 Thread-17上厕所
 Thread-11上完厕所走了
 Thread-16上完厕所走了
 Thread-10上完厕所走了
 Thread-18上厕所
 Thread-19上厕所
 Thread-20上厕所
 Thread-18上完厕所走了
 Thread-20上完厕所走了
 Thread-15上完厕所走了
 Thread-19上完厕所走了
 Thread-17上完厕所走了

Process finished with exit code 0

6.Event

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

from threading import Event
Event.isSet() #返回event的状态值
Event.wait() #如果 event.isSet()==False将阻塞线程；
Event.set() #设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
Event.clear() #恢复

例如1.，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

from threading import Thread,Event,currentThread
import  time
e=Event()
def conn_mysql():
    #链接数据库
    count=1
    while not e.is_set(): #当没有检测到时候
        if count>3:
            raise ConnectionError(‘尝试链接的次数过多‘)
        print(‘\033[45m%s 第%s次尝试‘ % (currentThread(), count))
        e.wait(timeout=1)#等待检测（里面的参数是超时1秒）
        count += 1
    print(‘\033[44m%s 开始链接...‘ % (currentThread().getName()))

def  chek_mysql():
  #检测数据库
   print( print(‘\033[42m%s 检测mysql...‘ % (currentThread().getName())))
   time.sleep(5)
   e.set()

if __name__=="__main__":
    for i in range(3):
        t=Thread(target=conn_mysql)
        t.start()
    #e.set()
    t=Thread(target=chek_mysql)
    t.start()

def traffic_ligths():
    #红绿灯
    time.sleep(5)
    e.set()

def car():
    #车
    print(‘\033[42m %s 等绿灯\033[0m‘ % currentThread().getName())
    e.wait()
    print(‘\033[44m %s 车开始通行‘ % currentThread().getName())

if __name__==‘__main__‘:
    for i in range(10):
        t=Thread(target=car) #10辆车
        t.start()
    traffic_thread=Thread(target=traffic_ligths)
    traffic_thread.start()

7.定时器（Timer）

from threading import Timer
def func(n):
    print(‘hello world‘,n)
t=Timer(3,func,args=(123,))#等待三秒后执行func函数，因为func函数有参数，那就再传一个参数进去
t.start()

8.线程queue

queue队列 ：使用import queue，用法与进程Queue一样

queue.Queue(maxsize=0) #先进先出

import queue
q=queue.Queue(3)#先进先出
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)
print(q.get())
print(q.get())
print(q.get())

queue.``LifoQueue(maxsize=0)#先进后出

q=queue.LifoQueue()#先进后出
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)
print(q.get())
print(q.get())
print(q.get())

queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列

# put进入一个元组，元组的第一个元素是优先级
# （通常也可以是数字，或者也可以是非数字之间的比较）
# 数字越小，优先级越高
import queue
q=queue.PriorityQueue()
q.put((20,‘a‘))
q.put((10,‘b‘))
q.put((30,‘c‘))
print(q.get())
print(q.get())
print(q.get())

https://www.cnblogs.com/lanyinhao/p/9223301.html

Python中的线程和进程

标签：mon   运行   RoCE   完成后   input   进程和线程   space   cti   owa   

原文地址：https://www.cnblogs.com/geoffreygao/p/13356813.html