多线程通信方式

共享变量

创建全局变量，多个线程公用一个全局变量，方便简单。但是坏处就是共享变量容易出现数据竞争，不是线程安全的，解决方法就是使用互斥锁。

# 示例代码，模拟爬虫

import threading

import time

url_lists = []

def get_urls():

# 模拟爬取url

global url_lists

print("get urls started")

for i in range(20):

url_lists.append(f"https://www.baidu.com/{i}")

print("get urls end")

def get_detail():

# 模拟爬取页面内容

global url_lists

if len(url_lists):

url = url_lists.pop()

print("get detail started")

time.sleep(2)

print("get detail end")

if __name__ == '__main__':

# 爬取url链接

thread_get_urls = threading.Thread(target=get_urls)

thread_get_urls.start()

# 开启10个线程爬取

for i in range(10):

t = threading.Thread(target=get_detail)

t.start()

将共享变量以参数传递进去

import threading

import time

url_lists = []

def get_urls(url_lists):

# 模拟爬取url

while True:

print("get urls started")

for i in range(20):

url_lists.append(f"https://www.baidu.com/{i}")

print("get urls end")

def get_detail(url_lists):

# 模拟爬取页面内容

while True:

if len(url_lists):

url = url_lists.pop()

print("get detail started")

time.sleep(2)

print("get detail end")

if __name__ == '__main__':

# 爬取url链接

thread_get_urls = threading.Thread(target=get_urls, args=(url_lists, ))

thread_get_urls.start()

# 开启10个线程爬取

for i in range(10):

t = threading.Thread(target=get_detail, args=(url_lists, ))

t.start()

将共享变量单独放在其他py文件中，应用场景变量很多时能方便管理，比如放在variables.py中

import threading

import time

import variables

def get_urls():

# 模拟爬取url

url_lists = variables.url_lists

while True:

print("get urls started")

for i in range(20):

url_lists.append(f"https://www.baidu.com/{i}")

print("get urls end")

def get_detail():

# 模拟爬取页面内容

url_lists = variables.url_lists

while True:

if len(url_lists):

url = url_lists.pop()

print("get detail started")

time.sleep(2)

print("get detail end")

if __name__ == '__main__':

# 爬取url链接

thread_get_urls = threading.Thread(target=get_urls)

thread_get_urls.start()

# 开启10个线程爬取

for i in range(10):

t = threading.Thread(target=get_detail)

t.start()

队列

线程间使用队列进行通信，因为队列所有方法都是线程安全的，所以不会出现线程竞争资源的情况。Queue常用的方法。

put(item, block=True, timeout=None)

阻塞方式将item添加进队列中，如果队列满了则一直等待，如果给定了timeout则等待timeout；如果block为Flase，则为非阻塞式，队列满时再添加则直接抛出错误

put_nowait(item)

非阻塞式添加

get(block=True, timeout=None)

阻塞式获取，队列为空时，则一直等待，或者等待给定timeout秒

get_nowait()

非阻塞式获取值

qsize()

返回队列大小

empty()

返回布尔值，判断队列是否为空

full()

返回布尔值，判断队列是否满了

join()

一直阻塞直到队列中的所有项目都已获取并处理完毕。

每当任务(示例：未爬取过的url)添加到队列时，未完成任务的计数就会增加。 每当消费者线程(示例：爬取网页内容的函数)调用task_done()以指示检索到该项目并且其上的所有工作都已完成时，计数就会下降。 当未完成任务的数量降至零时，join()取消阻塞

task_done()

表明以前排队的任务(示例：使用一个url爬取网页内容完成)已完成。

由队列使用者线程使用。每次调用get()方法从队列中获取任务，如果任务处理完毕，则条用task_done()方法，告知等待的队列(queue.join()这里在等待)任务的处理已完成。

如果join()当前正在阻塞，则它将在所有项目都已处理后恢复(这意味着已为每个已放入队列的项目收到task_done()调用)。

如果调用的次数超过队列中放置的项目，则引发ValueError。

最后两个方法，是我开始最不能理解的，后面看了很多博客，大概知道她们的作用。下面我以生产者消费者示例代码演示。代码从另外个哥们那里获取的，但是做了些修改

from threading import Thread

import time

import random

from queue import Queue

from collections import deque

# 创建队列，设置队列最大数限制为3个

queue = Queue(3)

# 生产者线程

class Pro_Thread(Thread):

def run(self):

# 原材料准备，等待被生产

tasks = deque([1, 2, 3, 4, 5, 6, 7, 8])

global queue

while True:

try:

# 从原材料左边开始生产，如果tasks中没有元素，调用popleft()则会抛出错误

task = tasks.popleft()

queue.put(task)

print("生产", task, "现在队列数:", queue.qsize())

# 休眠随机时间

time.sleep(random.random())

# 如果原材料被生产完，生产线程跳出循环

except IndexError:

print("原材料已被生产完毕")

break

print("生产完毕")

# 消费者线程

class Con_Thread(Thread):

def run(self):

global queue

while True:

if not queue.empty():

# 通过get(),这里已经将队列减去了1

task = queue.get()

time.sleep(2)

# 发出完成的信号，不发的话，join会永远阻塞，程序不会停止

queue.task_done()

print("消费", task)

else:

break

print("消费完毕")

# r入口方法，主线程

def main():

Pro_1 = Pro_Thread()

# 启动线程

Pro_1.start()

# 这里休眠一秒钟，等到队列有值，否则队列创建时是空的，主线程直接就结束了，实验失败，造成误导

time.sleep(1)

for i in range(2):

Con_i = Con_Thread()

# 启动线程

Con_i.start()

global queue

# 接收信号，主线程在这里等待队列被处理完毕后再做下一步

queue.join()

# 给个标示，表示主线程已经结束

print("主线程结束")

if __name__ == '__main__':

main()

threading.Thread().join()方法和queue.join)()的区别

线程的join()是主线程等待子线程的执行完毕再执行

队列的join()是主线程等待队列中的任务都消耗完再执行