python并发编程之多线程(实践篇)

Posted jiangfan95

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python并发编程之多线程(实践篇)相关的知识,希望对你有一定的参考价值。

一.threading模块介绍

官网链接:https://docs.python.org/3/library/threading.html?highlight=threading#

1.开启线程的两种方式

#直接调用
import threading
import time
def run(n):
    print(task,n)
     time.sleep(2)

t1 = threading.Thread(target=run,args=(t1,))
t1.start()
#继承式调用
mport threading
import time
class MyThread(threading.Thread):
    def __init__(self,n,sleep_time):
        super(MyThread, self).__init__()
        self.n = n
        self.sleep_time = sleep_time

    def run(self):
        print(running task,self.n)
        time.sleep(self.sleep_time)
        print(task done,,self.n)

t1 = MyThread(t1,2)
t1.start()

2.在一个进程下开启多个线程与在一个进程下开启多个子进程的区别

技术图片
from threading import Thread
from multiprocessing import Process
import os

def work():
    print(hello)

if __name__ == __main__:
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()
    print(主线程/主进程)
    ‘‘‘
    打印结果:
    hello
    主线程/主进程
    ‘‘‘

    #在主进程下开启子进程
    t=Process(target=work)
    t.start()
    print(主线程/主进程)
    ‘‘‘
    打印结果:
    主线程/主进程
    hello
    ‘‘‘
1.开启速度比较
技术图片
from threading import Thread
from multiprocessing import Process
import os

def work():
    print(hello,os.getpid())

if __name__ == __main__:
    #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print(主线程/主进程pid,os.getpid())

    #part2:开多个进程,每个进程都有不同的pid
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print(主线程/主进程pid,os.getpid())
2.比较pid
技术图片
from  threading import Thread
from multiprocessing import Process
import os
def work():
    global n
    n=0

if __name__ == __main__:
    # n=100
    # p=Process(target=work)
    # p.start()
    # p.join()
    # print(‘主‘,n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100


    n=1
    t=Thread(target=work)
    t.start()
    t.join()
    print(,n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据
3.数据是否共享

3.应用

1)将socket通信改写为多线程模式

技术图片
#_*_coding:utf-8_*_
#!/usr/bin/env python
import multiprocessing
import threading

import socket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind((127.0.0.1,8080))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print(data)
        conn.send(data.upper())

if __name__ == __main__:

    while True:
        conn,addr=s.accept()
        p=threading.Thread(target=action,args=(conn,))
        p.start()
多线程并发的socket服务端
技术图片
#_*_coding:utf-8_*_
#!/usr/bin/env python


import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect((127.0.0.1,8080))

while True:
    msg=input(>>: ).strip()
    if not msg:continue

    s.send(msg.encode(utf-8))
    data=s.recv(1024)
    print(data)
客户端

2)三个任务,一个接收用户输入,一个将用户输入的内容格式化成大写,一个将格式化后的结果存入文件

技术图片
rom threading import Thread
msg_l=[]
format_l=[]
def talk():
    while True:
        msg=input(>>: ).strip()
        if not msg:continue
        msg_l.append(msg)

def format_msg():
    while True:
        if msg_l:
            res=msg_l.pop()
            format_l.append(res.upper())

def save():
    while True:
        if format_l:
            with open(db.txt,a,encoding=utf-8) as f:
                res=format_l.pop()
                f.write(%s\\n %res)

if __name__ == __main__:
    t1=Thread(target=talk)
    t2=Thread(target=format_msg)
    t3=Thread(target=save)
    t1.start()
    t2.start()
    t3.start()
View Code

3)主线程等待子线程结束

技术图片
from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(%s say hello %name)

if __name__ == __main__:
    t=Thread(target=sayhi,args=(egon,))
    t.start()
    t.join()    #主线程等待子线程运行结束了再往下走
    print(主线程)
    print(t.is_alive())
    ‘‘‘
    egon say hello
    主线程
    False
    ‘‘‘
join()方法

 

二.守护线程

无论是进程还是线程,都遵循:守护xxx会等待主xxx运行完毕后被销毁

1)对主进程来说,运行完毕指的是主进程代码运行完毕

2)对主线程来说,运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕,主线程才算运行完毕

需要强调的是:运行完毕并非终止运行

技术图片
from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(%s say hello %name)

if __name__ == __main__:
    t=Thread(target=sayhi,args=(egon,))
    t.setDaemon(True) #必须在t.start()之前设置
    t.start()

    print(主线程)
    print(t.is_alive()) #结果为True说明此时主线程并没结束,守护进程还在
    ‘‘‘
    主线程
    True
    ‘‘‘
守护线程生命周期
技术图片
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(3)
    print("end123")

def bar():
    print(456)
    time.sleep(1)
    print("end456")


t1=Thread(target=foo)
t2=Thread(target=bar)

t1.daemon=True  #将t1设置为守护进程,主进程结束后t1也结束,
t1.start()  #可能会出现t1没有完全完全走完就结束的情况
t2.start()
print("main-------")

"""
运行结果:
123
456
main-------
end456
"""
案例分析

三.Python GIL(Global Interpreter Lock)

https://www.cnblogs.com/linhaifeng/articles/7449853.html

五.同步锁

1.GIL与lock

1)线程抢的是GIL锁,GIL锁相当于执行权限,拿到执行权限后才能拿到互斥锁Lock,其他线程也可以抢到GIL,但如果发现Lock仍然没有被释放则阻塞,即便是拿到执行权限GIL也要立刻交出来

2)join是等待所有,即整体串行,而锁只是锁住修改共享数据的部分,即部分串行,要想保证数据安全的根本原理在于让并发变成串行,join与互斥锁都可以实现,毫无疑问,互斥锁的部分串行效率要更高

3)GIL 与Lock是两把锁,保护的数据不一样,前者是解释器级别的(当然保护的就是解释器级别的数据,比如垃圾回收的数据),后者是保护用户自己开发的应用程序的数据,很明显GIL不负责这件事,只能用户自定义加锁处理,即Lock

2.过程分析

所有线程抢的是GIL锁,或者说所有线程抢的是执行权限

线程1抢到GIL锁,拿到执行权限,开始执行,然后加了一把Lock,还没有执行完毕,即线程1还未释放Lock,有可能线程2抢到GIL锁,开始执行,执行过程中发现Lock还没有被线程1释放,于是线程2进入阻塞,被夺走执行权限,有可能线程1拿到GIL,然后正常执行到释放Lock。。。这就导致了串行运行的效果

既然是串行,那我们执行

t1.start()

t1.join

t2.start()

t2.join()

这也是串行执行啊,为何还要加Lock呢,需知join是等待t1所有的代码执行完,相当于锁住了t1的所有代码,而Lock只是锁住一部分操作共享数据的代码。

3.Lock使用

锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象,当你需要访问该资源时,调用acquire方法来获取锁对象(如果其它线程已经获得了该锁,则当前线程需等待其被释放),待资源访问完后,再调用release方法释放锁:

import threading

R=threading.Lock()

R.acquire()  #获取所对象
‘‘‘
对公共数据的操作
‘‘‘
R.release()  #释放

 

技术图片
#1.100个线程去抢GIL锁,即抢执行权限
#2. 肯定有一个线程先抢到GIL(暂且称为线程1),然后开始执行,一旦执行就会拿到lock.acquire()
#3. 极有可能线程1还未运行完毕,就有另外一个线程2抢到GIL,然后开始运行,但线程2发现互斥锁lock还未被线程1释放,于是阻塞,被迫交出执行权限,即释放GIL
#4.直到线程1重新抢到GIL,开始从上次暂停的位置继续执行,直到正常释放互斥锁lock,然后其他的线程再重复2 3 4的过程
GIL锁与互斥锁综合分析
技术图片
#不加锁:并发执行,速度快,数据不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print(%s is running %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == __main__:
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print(主:%s n:%s %(stop_time-start_time,n))

‘‘‘
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
‘‘‘


#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    #未加锁的代码并发运行
    time.sleep(3)
    print(%s start to run %current_thread().getName())
    global n
    #加锁的代码串行运行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == __main__:
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print(主:%s n:%s %(stop_time-start_time,n))

‘‘‘
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
‘‘‘

#思考:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print(%s start to run %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == __main__:
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print(主:%s n:%s %(stop_time-start_time,n))

‘‘‘
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
‘‘‘
互斥锁与join的区别

 

六.死锁现象与递归锁

所谓死锁: 是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁

技术图片
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(10):
        t=MyThread()
        t.start()

‘‘‘
Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁
然后就卡住,死锁了
‘‘‘
死锁现象

解决方法,递归锁,在Python中为了支持在同一线程中多次请求同一资源,python提供了可重入锁RLock。

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

mutexA=mutexB=threading.RLock() #一个线程拿到锁,counter加1,该线程内又碰到加锁的情况,则counter继续加1,
#这期间所有其他线程都只能等待,等待该线程释放所有锁,即counter递减到0为止

 

七.信号量Semaphore

同进程的一样

Semaphore管理一个内置的计数器,
每当调用acquire()时内置计数器-1;
调用release() 时内置计数器+1;
计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。

实例:(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5):

技术图片
from threading import Thread,Semaphore
import threading
import time
# def func():
#     if sm.acquire():
#         print (threading.currentThread().getName() + ‘ get semaphore‘)
#         time.sleep(2)
#         sm.release()
def func():
    sm.acquire()
    print(%s get sm %threading.current_thread().getName())
    time.sleep(3)
    sm.release()
if __name__ == __main__:
    sm=Semaphore(5)
    for i in range(23):
        t=Thread(target=func)
        t.start()
View Code

与进程池是完全不同的概念,进程池Pool(4),最大只能产生4个进程,而且从头到尾都只是这四个进程,不会产生新的,而信号量是产生一堆线程/进程

 互斥锁与信号量推荐博客:http://url.cn/5DMsS9r

八.Event

同进程的一样

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

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

event.wait():如果 event.isSet()==False将阻塞线程;

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

event.clear():恢复event的状态值为False。
技术图片
import threading,time
event = threading.Event()
def lighter():
    count = 0
    event.set() #先设置绿灯
    while True:
        if count > 5 and count < 10:#改成红灯
            event.clear()#标志位清了
            print(\\033[41;1mred light is on ...\\033[0m)
        elif count > 10:
            event.set()#变绿灯
            count = 0
        else:
            print(\\033[42;1mgreen light is on ...\\033[0m)
        time.sleep(1)
        count += 1

def car(name):
    while True:
        if event.is_set():#代表绿灯
            print([%s] running...%name)
            time.sleep(1)
        else:
            print([%s] sees red light ,waiting ... %name)
            event.wait()
            print(\\033[34;1m[%s] green light is on,start going ... \\033[0m %name)


light = threading.Thread(target=lighter,)
light.start()

car1 = threading.Thread(target=car,args=(宝马,))
car1.start()
红绿灯

 

九.条件Condition

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

import threading
 
def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()
 
if __name__ == __main__:
 
    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()
 
    while True:
        inp = input(>>>)
        if inp == q:
            break
        con.acquire()
        con.notify(int(inp))
        con.release()

 

十.定时器

定时器,指定n秒后执行某操作

 

from threading import Timer
def hello():
    print("hello, world")
 
t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

 

技术图片
from threading import Timer
import random,time

class Code:
    def __init__(self):
        self.make_cache()

    def make_cache(self,interval=5):
        self.cache=self.make_code()
        print(self.cache)
        self.t=Timer(interval,self.make_cache)
        self.t.start()

    def make_code(self,n=4):
        res=‘‘
        for i in range(n):
            s1=str(random.randint(0,9))
            s2=chr(random.randint(65,90))
            res+=random.choice([s1,s2])
        return res

    def check(self):
        while True:
            inp=input(>>: ).strip()
            if inp.upper() ==  self.cache:
                print(验证成功,end=\\n)
                self.t.cancel()
                break


if __name__ == __main__:
    obj=Code()
    obj.check()
验证码定时器

 

 

十一.线程queue

queue队列 :使用import queue,用法与进程Queue一样

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

 

技术图片
import queue

q=queue.Queue()
q.put(first)
q.put(second)
q.put(third)

print(q.get())
print(q.get())
print(q.get())
‘‘‘
结果(先进先出):
first
second
third
‘‘‘
View Code

 

class queue.LifoQueue(maxsize=0) #last in fisrt out 

技术图片
import queue

q=queue.LifoQueue()
q.put(first)
q.put(second)
q.put(third)

print(q.get())
print(q.get())
print(q.get())
‘‘‘
结果(后进先出):
third
second
first
‘‘‘
后进先出

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

技术图片
import queue

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

print(q.get())
print(q.get())
print(q.get())
‘‘‘
结果(数字越小优先级越高,优先级高的优先出队):
(10, ‘b‘)
(20, ‘a‘)
(30, ‘c‘)
‘‘‘
设置优先级

 

 

 

 

以上是关于python并发编程之多线程(实践篇)的主要内容,如果未能解决你的问题,请参考以下文章

Python并发编程之多线程

python并发编程之多线程编程

并发编程路线

python并发编程之多线程

python并发编程之多线程

python并发编程之多线程