Python的多线程使用实践

发布时间:2024年01月15日

这是在一个小的爬虫项目中使用到的线程创建的方式,比较简单。

t1 = threading.Thread(target=start_gh(), name='t1')
t1.start()

主要是通过threading模块进行线程创建,那么有没有类似Java线程池的概念呢,我想是有的,接下来探索下Python多线程。

自定义线程类

上面使用threading创建线程,还可以使用继承Thread的方式创建线程类,有点类似Java实现Runnable接口

import threading

class MyThreading(threading.Thread):

    def __init__(self, func, arg):
        super(MyThreading,self).__init__()
        self.func = func
        self.arg = arg

    def run(self):
        self.func(self.arg)

Python锁

有多线程,就必有锁,python有以下几种锁。

  • Lock 互斥锁
  • RLock 可重入锁
  • Semaphore 信号
  • Event 事件
  • Condition 条件
  • Barrier “阻碍”
number = 0
lock = threading.Lock()

def plus(lk: threading.Lock):
    global number
    lk.acquire()
    for _ in range(100000000):
        number = number + 1
    print(f"{threading.currentThread().getName()} result:{number}")
    lk.release()

if __name__ == '__main__':
    for i in range(2):  # 用2个子线程,就可以观察到脏数据
        t = threading.Thread(target=plus, args=(lock,), name=f"t{i}")  # 需要把锁当做参数传递给plus函数
        t.start()
    time.sleep(2)  # 等待2秒,确保2个子线程都已经结束运算。
    print("主线程执行完毕后,number = ", number)

上面手动acquire和release太麻烦?使用with语句

 global number
	  with lk:
	    for _ in range(100000000):
	        number = number + 1
	    print(f"{threading.currentThread().getName()} result:{number}")

定时器timer

指定n秒后操作,这个小功能还是很好用的

from threading import Timer

def hello():
    print("hello, world")

# 表示1秒后执行hello函数
t = Timer(1, hello)
t.start()

python线程池使用

使用python自带的ThreadPoolExecutor模块

from concurrent.futures import ThreadPoolExecutor
import time

# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

executor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))
# done方法用于判定某个任务是否完成
print(task1.done())
# cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
print(task2.cancel())
time.sleep(4)
print(task1.done())
# result方法可以获取task的执行结果
print(task1.result())

# 执行结果
# False  # 表明task1未执行完成
# False  # 表明task2取消失败,因为已经放入了线程池中
# get page 2s finished
# get page 3s finished
# True  # 由于在get page 3s finished之后才打印,所以此时task1必然完成了
# 3     # 得到task1的任务返回值

as_completed

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]

for future in as_completed(all_task):
    data = future.result()
    print("in main: get page {}s success".format(data))

# 执行结果
# get page 2s finished
# in main: get page 2s success
# get page 3s finished
# in main: get page 3s success
# get page 4s finished
# in main: get page 4s success

as_completed()方法是一个生成器,在没有任务完成的时候,会阻塞,在有某个任务完成的时候,会yield这个任务,就能执行for循环下面的语句,然后继续阻塞住,循环到所有的任务结束。从结果也可以看出,先完成的任务会先通知主线程

wait

wait方法可以让主线程阻塞,直到满足设定的要求。

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time

# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果 
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main
文章来源:https://blog.csdn.net/Kingsea442/article/details/135606966
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。