Python多线程----线程池以及线程实现异步任务

Python多线程----线程池

需求：假设我们现在有一个多线程项目，每有一个用户连接进来，我们的服务器就会创建一个线程。而我们的服务器最多能够承载100个线程，再多就会崩溃。为了防止恶意用户伪装真实用户构建大量的访问来让我们的服务器崩溃，现在需要对线程数量进行限制，一共只有100个线程，并且当一个用户访问结束以后线程会自动归还，等待下一个用户访问。如果100个线程全部被占用则101个用户进入阻塞时间，直到某一个用户退出，线程得到释放，101个用户才能被通行。

不难看出上面的需求，类似我们MySQL的连接池。既然如此，我们的Python也应该有一个线程池，并且这种问题非常的常见，肯定已经有现成的库以供我们使用。今天我们就来看一下Python标准库中from concurrent.futures下的ThreadPoolExecutor。

第一个例子

# 首先导包
from concurrent.futures import ThreadPoolExecutor

# 创建线程池
executor = ThreadPoolExecutor(10)

# 测试方法
def test_function(num1, num2):
    print(num1, num2)
    return num1 + num2

# 第一个参数为具体的方法，后面为方法的参数
future = executor.submit(test_function, 1, 2)
# future的result()方法可以获取到函数的执行结果
print(future.result())

执行结果：

1 2
3

ThreadPoolExecutor(pool_count): pool_count代表创建线程的数量，会返回一个该线程池的执行者对象，这个对象的submit()方法和map()方法，能够使用线程池中的线程来执行我们指定的方法，并且返回一个Future对象。Future对象的result()方法，可以获取我们方法执行的结果。如果方法一直没有返回或执行完毕，则result()方法会进入阻塞状态，直到我们的方法返回或执行完毕。

使用map()方法批量执行

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(10)

def test_function(num1, num2):
    print(num1, num2)
    return num1 + num2

"""
    executor.map(function, 参数1_list, 参数2_list, 参数n_list)
    参数1_list: 代表方法第一个参数的列表
    参数2_list: 代表方法第二个参数的列表
    如：
        executor.map(test_function, [1, 2], [5, 5])
        代表，执行test_function方法，第一个线程的参数为1和5，第二个线程的参数为2和5。
        线程1：test_function(1, 5) 结果为1 + 5 = 6
    该方法返回的是一个可迭代的对象，里面直接包含了每个方法执行的结果，不需要调用result()方法。
    详情：https://docs.python.org/3/library/concurrent.futures.html
"""
result_iterators = executor.map(test_function, [1, 2], [5, 5])

for result in result_iterators:
    print(result)

执行结果：

1 5
2 5
6
7

尝试一下所有线程都被占用的情况

import time
from concurrent.futures import ThreadPoolExecutor

# 方便测试创建三个线程
executor = ThreadPoolExecutor(3)

def test_function(num1, num2):
    print(num1, num2)
    # 方法休眠十秒
    time.sleep(10)
    return num1 + num2

# 使用三个线程，占用线程池全部线程
# 由于我们的结果是十秒后返回，所以这里也会被阻塞，十秒后才会收到结果
result_iterators = executor.map(test_function, [1, 2, 3], [5, 6, 7])

for result in result_iterators:
    print(result)

# 到这里很显然前面三个线程都在使用中，10秒后才能得到执行
future = executor.submit(test_function, 4, 8)
print(future.result())

执行结果：

1 5
2 6
3 7
6
8
10
4 8
12
[Finished in 20.2s]

https://www.imooc.com/article/16217

 1 import  threading,time
 2 
 3 def thead(num):
 4     time.sleep(1)
 5     print("阻塞程序%s开始执行"%num)
 6     time.sleep(3)
 7     print("阻塞程序%s执行完毕"%num)
 8 
 9 def main():
10     print("主方法开始执行")
11 
12     for i in range(1,3):
13         thead(i)
14 
15     print("主方法执行完毕")
16     return
17 
18 if __name__ == '__main__':
19     print(time.ctime())
20     num = main()
21     print("返回结果为%s"%num)
22     print(time.ctime())

Wed Nov 21 09:22:56 2018
主方法开始执行
阻塞程序1开始执行
阻塞程序1执行完毕
阻塞程序2开始执行
阻塞程序2执行完毕
主方法执行完毕
返回结果为None
Wed Nov 21 09:23:04 2018

import  threading,time

def thead(num):
    # time.sleep(1)
    print("线程%s开始执行"%num)
    time.sleep(3)
    print("线程%s执行完毕"%num)

def main():
    print("主方法开始执行")

    #创建2个线程
    poll = []#线程池
    for i in range(1,3):
        thead_one = threading.Thread(target=thead, args=(i,))
        poll.append(thead_one) #线程池添加线程
    for n in poll:
        n.start()   #准备就绪,等待cpu执行

    print("主方法执行完毕")
    return

if __name__ == '__main__':
    print(time.ctime())
    num = main()
    print("返回结果为%s"%num)
    print(time.ctime())

from flask import Flask
import time
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()
app = Flask(__name__)


@app.route('/')
def update_redis():
    executor.submit(do_update)
    return 'ok'


def do_update():
    time.sleep(3)
    print('start update cache')
    time.sleep(1)
    print("end")


if __name__ == '__main__':
    app.run(debug=True)