2022年 11月 3日

python3之多线程

多线程简介

我们进行程序开发的时候，肯定避免不了要处理并发的情况，一般并发的手段有采用多进程和多线程，但线程比进程更轻量化，系统开销一般也更低，所以大家更倾向于用多线程的方式处理并发的情况。

python3 中多线程使用threading模块中的Thread类来实现多线程并发的，threading为python3标准库中的模块，无需安装，直接导入使用即可。

Thread类

Thread类的主要参数如下所示，仅需重点关注target与args两个参数即可，target参数表示要传入的函数名，args参数则是出入函数需要使用的参数，格式必须为元组，不然报错。

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
  • 1

对上面的常用参数的解释

  • target 传入要运行的函数名
  • name 线程名称，默认格式为Thread-N,N为数字
  • args 传入需要的参数值，格式为元组，最后以“,”结尾（a,b,）
  • daemon 属性默认是 False

Thread类基本使用方法

python 中使用多线程有两种方式，一种是创建threading.Thread的实例对象，传入需要执行的函数和参数来使用，另一种则是继承Thread类，重写run()方法来确定执行的内容。
创建Thread对象实例来执行的示例
Thread类中常用的函数

  • start(),启动线程
  • join() 阻塞调用线程，个人实验中以为，join()线程调用该函数之后，将在执行完该线程之后继续主线程，否则主线程会与启动的子线程同步执行
  • isAlive()，线程是否存活
  • getName()，获取线程名称
  • setName(),设置线程名称

join 函数作用的实验

实验时调用join函数，则done均为最后输出，若注实掉join函数的两行代码则done在中间便会输出。

import threading

def pr1():
    for i in range(100):
        print("pr1 " + str(i))

def pr2():
    for i in range(100):
        print("pr2 " + str(i))

thread1 = threading.Thread(target=pr1)
thread2 = threading.Thread(target=pr2)
thread1.start()
thread2.start()

thread1.join()   
thread2.join()

print("done")
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

Thread模块使用

直接使用

from threading import Thread


def test(value):
    print(value)


for i in range(10):
    a = Thread(target=test, args=(i,))  #建议这样写，以免报错
    a.start()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

继承Thread类来写

import threading


class MyThread(threading.Thread):
    def __init__(self, name=None):   #继承的写法要注意这个函数的写法
        threading.Thread.__init__(self, name=name)

    def run(self):
        test()


def test():
    for i in range(10):
        print(i)


a = MyThread(name="a")
a.start()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

锁 threading.Lock()

如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。既保证每次只有一个线程对某个数据进行修改，在实际输入输出、连接数据库插入数据时个人曾经遇到过格式混乱、内容错误的问题，想来也是由于多个线程同时执行某个操作导致的。

其实单个线程多次调用I/O等操作时不怕，就怕多个线程同时进行调用，则会混乱，若是插入数据库时存在这样的混乱则数据库很难看，可能会导致数据没有参考价值。

未使用锁时

import threading


def pr1():
    for i in range(10):
        print("pr1 " + str(i))


def pr2():
    for i in range(10):
        print("pr2 " + str(i))


thread1 = threading.Thread(target=pr1)
thread2 = threading.Thread(target=pr2)
thread1.start()
thread2.start()

thread1.join()
thread2.join()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

不使用锁时，输出很乱，若是插入数据库时存在这样的混乱则数据库很难看，可能会导致数据没有参考价值。
在这里插入图片描述

使用锁时

import threading
lock = threading.Lock()


def pr1():
    lock.acquire()    #输出位置加锁，输出美观的一批
    for i in range(10):
        print("pr1 " + str(i))
    lock.release()


def pr2():
    lock.acquire()
    for i in range(10):
        print("pr2 " + str(i))
    lock.release()


thread1 = threading.Thread(target=pr1)
thread2 = threading.Thread(target=pr2)
thread1.start()
thread2.start()

thread1.join()
thread2.join()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25

美观的一批
在这里插入图片描述

另一种写法，with 语句会在这个代码块执行前自动获取锁，在执行结束后自动释放锁。

import threading

lock = threading.Lock()
with lock:
    # 这里写自己的代码
    pass
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

Queue 队列

queue模块中有多种对列，但是平时用到只需要Queue队列即可，队列是线程安全的类，在多个线程同时访问时不会出现问题，而dict等类型（不是线程安全的类型）则不行，下面的代码若num为dict类型，则会多次循环打印，Queue类型则会只打印一次。

queue.Queue(maxsize=0)
先进先出(First In First Out: FIFO)队列，最早进入队列的数据拥有出队列的优先权，就像看电影入场时排队一样，排在队伍前头的优先进入电影院。入参 maxsize 是一个整数，用于设置队列的最大长度。一旦队列达到上限，插入数据将会被阻塞，直到有数据出队列之后才可以继续插入。如果 maxsize 设置为小于或等于零，则队列的长度没有限制。

Queue中的常用函数

  • qsize()，返回队列大小
  • empty() 如果队列为空，返回True,反之False
  • get() 获取对列的值
  • put() 向队列中添加数值。
  • Queue.task_done()在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
  • Queue.join()实际上意味着等到队列为空，再执行别的操作

task_done() 和join()函数的理解

如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，在最后执行个join()是等不到结果的，会一直挂起。
可以理解为，每task_done一次 就从队列里删掉一个元素，这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。

Thread 类结合 Queue队列，可以理解为生产者，消费者问题，Queue为生产者，而三个线程为消费者（线程之间数据共享，共同处理Queue中的数据）

import threading
import queue

num = queue.Queue()

for i in range(10):
    num.put(i)


def aaa(a):   //打印a队列，并将a队列的值
    while not a.empty():
        print(a.get())    
    

//线程之间，数据共享，共同对Queue进行输出，只循环输出一次
pr1 = threading.Thread(target=aaa, args=(num,))
pr2 = threading.Thread(target=aaa, args=(num,))
pr3 = threading.Thread(target=aaa, args=(num,))

pr1.start()
pr2.start()
pr3.start()

pr1.join()
pr2.join()
pr3.join()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

参考

https://www.runoob.com/python3/python3-multithreading.html
https://www.cnblogs.com/franknihao/p/6627857.html
https://cloud.tencent.com/developer/article/1383120
https://juejin.cn/post/6844903617535672328
https://www.runoob.com/python3/python3-multithreading.html
http://www.ityouknow.com/python/2019/10/10/python-queue-029.html