python 多线程知识全面解析

非阻塞启动线程

import threading
import time
def one_thread(name,id):
    print("start....")
    print(name)
    print(id)
    time.sleep(5)
    print("end...")

print("start thread")
threading.Thread(target=one_thread, args=(), kwargs={"name": 111, "id": 222}).start()
# args是一个list
# kwargs是一个字典，需要对应函数的key
print("end thread")

• 得到值如下，线程启动函数后，非阻塞执行

start thread
start....
111
222
end thread
end...

多线程并发处理

import threading
import time


class myThread(threading.Thread):
    def __init__(self, threadID, name):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name

    def run(self):
        print_time(self.threadID, self.name)

num = 0
def print_time(threadID, name):
    global num
    # 每一个线程循环10次，最终总循环次数为30次
    for i in range(10):
        print("start run")
        time.sleep(2)
        print(i)
        num += 1
    print("thread_id=%s:name=%s" % (threadID, name))


if __name__ == '__main__':
    threads = []
    # 新增三个线程
    for i in range(3):
        name = "Thread-%d" % i
        t = myThread(i, name)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    print("所有线程执行完毕")
    print("总循环次数为：%s" % num)

• 打印结果：每次运行三个线程，每个线程循环打印10次

start run
start run
start run
0
0
...
thread_id=1:name=Thread-1
所有线程执行完毕
总循环次数为：30

• 多线程共享资源，可以使用全局变量global

多线程加锁

• 对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间

# -*- coding: utf-8 -*-
import time
import threading
# 创建锁对象
lock = threading.Lock()
num = 0

def run(n):
    global num
    for i in range(10):
        # 加锁  为了确保下面代码只能由一个线程从头到尾的执行
        # 会阻止多线程的并发执行，所以效率会大大降低
        """
        lock.acquire()
        try:
            num = num - n
            num = num + n
        finally:
            # 解锁
            lock.release()
        """
        with lock:
            time.sleep(2)
            print("start")
            num = num + 1
            print("==============")


if __name__ == '__main__':
    t1 = threading.Thread(target=run,args=(6,))
    t2 = threading.Thread(target=run,args=(9,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("num = %s"%(num))

• 打印结果是每次只能运行一个线程

start
==============
...
num = 20

多线程与队列

• 我们经常会遇到这样的一个问题，这里有成千上万条数据，每次需要取出其中的一条数据进行处理，那么引入多线程该怎么进行任务分配？
• 我们可以将数据进行分割然后交给多个线程去跑，可是这并不是一个明智的做法。在这里我们可以使用队列与线程相结合的方式进行任务分配。
• 队列线程的思想： 首先创建一个全局共享的队列，队列中只存在有限个元素，并将所有的数据逐条加入到队列中，并调用队列的join函数进行等待。之后便可以开启若干线程，线程的任务就是不断的从队列中取数据进行处理就可以了。

import threading
import time
import queue

q = queue.Queue(10)

threadLock = threading.Lock()


class myThread(threading.Thread):
    def __init__(self, threadID, name):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.exitFlag = 0

    def run(self):
        while not self.exitFlag:
            threadLock.acquire()
            if not q.empty():
                id = q.get()
                print_time(self.name, id)
                threadLock.release()
            else:
                threadLock.release()


def print_time(threadName, id):
    print ("%s:%s:%s"%(threadName,time.ctime(time.time()),id))
    # pass


# 创建3个线程
threads = []
for i in range(3):
    name = "Thread-%d" % i
    t = myThread(i, name)
    t.start()
    threads.append(t)
print(threads)

# 新增队列数据
for i in range(10000):
    q_name = "Queue:%d" % i
    q.put(q_name)

# 等待队列清空
while not q.empty():
    pass

# 也可以join方法，与上同效
# q.join()

# 通知线程，处理完之后关闭
for t in threads:
    t.exitFlag = 1

# 等待所有线程结束之后才退出
for t in threads:
    t.join()

print("Exiting Main Thread")

• 这里必须要在判断q.empty()前加上线程锁，因为可能会出现这样的一种情况。
• 某一时刻，队列中还有一个元素，该元素正在被线程A取出，而与此同时线程B正在判断队列q是否为空，而此时线程B中队列q不为空进入后面的操作，但是待B去取元素时，最后一个元素已经被A取出，造成线程等待，显示出被挂起的状态。
• 我们也可以通过加入q.get(timeout=10)超时操作来弥补这一问题。
• 打印的结果

[<myThread(Thread-0, started 6568)>, <myThread(Thread-1, started 7724)>, <myThread(Thread-2, started 7796)>]
Thread-1:Sat Aug 22 11:36:29 2020:Queue:0
Thread-1:Sat Aug 22 11:36:29 2020:Queue:1
...
Thread-1:Sat Aug 22 11:36:30 2020:Queue:9998
Thread-1:Sat Aug 22 11:36:30 2020:Queue:9999
Exiting Main Thread

ThreadPoolExecutor线程池的使用

• 锁依然可以运用到线程池
• map的使用，接受一个List的数据，会循环调用

from concurrent.futures.thread import ThreadPoolExecutor
import time
num = 0
def print_time(data):
    global num
    num += 1
    time.sleep(2)
    print("start_%s" % data)
    print("============")
data = []
for i in range(50):
    data.append(i)
with ThreadPoolExecutor(10) as pool:
    result = pool.map(print_time, data)
# 等待所有线程执行完毕
for i in result:
    pass
print("循环次数=%s" % num)

• 打印结果为：每次启动10个线程，启动了5次

============
start_46
start_49
============
============
循环次数=50

• submit接受list的数据，也可以接受字典

rom concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures import as_completed

import time
def print_time(data):
    time.sleep(2)
    print("start_%s" % data)
    print("============")
data = []
for i in range(50):
    data.append(i)
with ThreadPoolExecutor(10) as executor:
    future_list = []
    for i in range(10):
        # future = executor.submit(print_time,data)
        future = executor.submit(print_time, {"name": 111, "id": 222})
        future_list.append(future)
    for res in as_completed(future_list):  # 这个futrure_list是你future对象的列表
        print(res.result())

参考

• https://www.jianshu.com/p/dd1b4077666e
• https://blog.csdn.net/vspiders/article/details/80724583