python多线程

Posted 2021-01-19 Wualin

多线程

• 线程之间的数据是共享的

如何开启线程

• threading模块（使用方法与multiprose一样）

import time
import random
from threading import Thread

def run(name):
    print(‘%s is running‘%name)
    time.sleep(random.randrange(1,5))
    print(‘%s is end‘%name)

if __name__ == ‘__main__‘:
    t1 = Thread(target=run,args=(‘喵‘,))
    t1.start()
    print(‘主线程‘)

import time
from threading import Thread
class MyThread(Thread):
    def __init__(self,name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s is running‘%self.name)
        time.sleep(2)
        print(‘%s is end‘%self.name)

if __name__ == ‘__main__‘:
    t1 = MyThread(‘喵‘)
    t1.start()
    print(‘主线程‘)

thread对象的其他属性与方法

from threading import Thread,currentThread,active_count,enumerate

import time

def task():
    print(‘%s is running‘%currentThread().getName())
    time.sleep(2)
    print(‘%s is done‘%currentThread().getName())

if __name__ == ‘__main__‘:
    t = Thread(target=task,name=‘子线程‘)
    t.start()
    #t.setName(‘儿子进程1‘)  设置线程名字
    print(t.isAlive())#判断线程是否存活
    print(‘主线程‘,currentThread().getName())
    # t.join()#等当前线程结束才继续执行主线程
    print(active_count())   #活跃的线程数
    print(enumerate())#[<_MainThread(MainThread, started 140735697388416)>, <Thread(子线程, started 123145519529984)>]

守护线程

• 在一个进程内，只有一个线程，线程运行结束，代表这个一个进程结束。
• 在一个进程内，开多个线程，主线程在代码运行完毕，还需要等待其他线程干完活才会结束

互斥锁

• 将并行编程串行，牺牲效率保证数据安全，与进程的互斥锁一样使用

GIL全局解释器锁

pass

死锁与递归锁

• 互斥锁只能acquire一次

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

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()

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

• 递归锁:可以acquire多次，每次acquire一次计数器+1，只要计数为0，才能被其他线程抢到

# 递归锁
from threading import Thread,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()

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

信号量：可以同时运行多个线程

from threading import Thread,Semaphore,currentThread
import time,random

sm = Semaphore(3)
def task():
    with sm:
        print(‘%s is run‘%currentThread().getName())
        time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    for i in range(10):
        t = Thread(target=task)
        t.start()

Event事件：实现线程同步

1. event.wait()#等待(可设置等待时间)
2. event.set()#开始
3. event.is_set()

from threading import Thread,Event
import time
event = Event()
def student(name):
    print(‘学生 %s正在听课‘%name)
    event.wait()
    print(‘学生%s下课了‘%name)

def teacher(name):
    print(‘老师%s 正在上课‘%name)
    time.sleep(7)
    print(‘老师%s 让学生下课了‘%name)
    event.set()

if __name__ == ‘__main__‘:
    s1 = Thread(target=student,args=(‘wualin‘,))
    s2 = Thread(target=student,args=(‘wxx‘,))
    s3 = Thread(target=student,args=(‘zxx‘,))
    t1 = Thread(target=teacher,args=(‘egon‘,))
    s1.start()
    s2.start()
    s3.start()
    t1.start()

定时器

线程queue

• 先进先出

import queue
q = queue.Queue(3) #先进先出->队列
q.put(5)
q.put(‘miao‘)
q.put(‘sta‘)

print(q.get())
print(q.get())
print(q.get())

#get 和 put可设置是否阻塞以及阻塞时间

print(q.get(block=True,timeout=3))

• 后进先出

q = queue.LifoQueue(3)#后进先出->堆栈

q.put(‘fisrt‘)
q.put(2)
q.put(‘miao‘)

print(q.get())
print(q.get())
print(q.get())

• 优先级队列

import queue
q = queue.PriorityQueue(3)#优先级队列
q.put((10,‘one‘))
q.put((40,‘two‘))
q.put((20,‘three‘))
#数字越小优先级越高
print(q.get())
print(q.get())
print(q.get())

进程池与线程池

• 池对线程或进程数量进行一个限制

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random

def task(name):
    print(‘name:%s pid:%s run‘%(name,os.getpid()))
    time.sleep(random.randint(1,3))


if __name__ == ‘__main__‘:
    pool = ProcessPoolExecutor(4)#进程池
    for i in range(10):
        pool.submit(task,‘egon%s‘%i)#异步调用

    pool.shutdown()#把往进程池提交任务的入口关闭

    print(‘主‘)

异步调用与回调机制

• 提交任务的两种方式

  • 同步调用：提交完任务后，就在原地等待任务执行完毕，拿到结果再执行下一行代码，程序是串行执行

  • 异步调用

    from concurrent.futures import ThreadPoolExecutor
import time,random

def eat(name):
    print(‘%s is eating‘%name)
    time.sleep(random.randint(2,5))
    res = random.randint(5,10)*‘#‘
    return {‘name‘:name,‘size‘:res}

def count(weith):
    weith = weith.result()#
    name = weith[‘name‘]
    size = len(weith[‘size‘])
    print(‘name:%s eat is %s‘%(name,size))

if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(5)
    pool.submit(eat,‘miao‘).add_done_callback(count)#回调机制
    pool.submit(eat,‘car‘).add_done_callback(count)
    pool.submit(eat,‘dog‘).add_done_callback(count)
    pool.submit(eat,‘zhang‘).add_done_callback(count)