快捷導(dǎo)航

Python常用模塊之threading和Thread模塊及線程通信

更新時(shí)間：2022年06月13日 09:33:03 作者：??孤寒者????

這篇文章主要介紹了Python常用模塊之threading和Thread模塊及線程通信，文章為圍繞主題的相關(guān)內(nèi)容展開(kāi)詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，需要的朋友看可以參考一下方法

1. 線程通信

1.1 互斥鎖

在多線程中 , 所有變量對(duì)于所有線程都是共享的 , 因此 , 線程之間共享數(shù)據(jù)最大的危險(xiǎn)在于多個(gè)線程同時(shí)修改一個(gè)變量 , 那就亂套了 , 所以我們需要互斥鎖 , 來(lái)鎖住數(shù)據(jù)。

1.2 線程間全局變量的共享

注意：

因?yàn)榫€程屬于同一個(gè)進(jìn)程，因此它們之間共享內(nèi)存區(qū)域。因此全局變量是公共的。

# -*- coding: utf-8 -*-
import threading
a = 1
def func():
    global  a
    a = 2
t = threading.Thread(target=func)
t.start()
t.join()
print(a)

1.3 共享內(nèi)存間存在競(jìng)爭(zhēng)問(wèn)題

先來(lái)個(gè)正常的例子，不用多線程：

# -*- coding: utf-8 -*-
x = 0
n =1000000
def a(n):
    global x
    for i in range(n):
        x += 1

def b(n):
    global x
    for i in range(n):
        x -= 1

a(n)
b(n)
print(x)

輸出肯定和大家想的一樣，毫無(wú)疑問(wèn)是0！

# -*- coding: utf-8 -*-
from threading import Thread

x = 0
n =1000000
def a(n):
    global x
    for i in range(n):
        x += 1

def b(n):
    global x
    for i in range(n):
        x -= 1

if __name__ == '__main__':
    a = Thread(target=a,args = (n,))
    b = Thread(target=b,args = (n,))
    a.start()
    b.start()
    # 一定要加阻塞，原因大家可以自己結(jié)合第一篇講的自己好好想想哦~
    a.join()
    b.join()
    print(x)

提示：

如果1000000不能出現(xiàn)效果可以繼續(xù)在后面加0

你會(huì)發(fā)現(xiàn)這個(gè)結(jié)果千奇百怪?。?！

1.4 使用鎖來(lái)控制共享資源的訪問(wèn)

下面引入互斥鎖

在多線程中 , 所有變量對(duì)于所有線程都是共享的 ,因此 ,線程之間共享數(shù)據(jù)最大的危險(xiǎn)在于多個(gè)線程同時(shí)修改一個(gè)變量 , 那就亂套了 , 所以我們需要互斥鎖 , 來(lái)鎖住數(shù)據(jù)。
只要我們操作全局變量的時(shí)候，就在操作之前加鎖，在操作完之后解鎖，就解決了這個(gè)資源競(jìng)爭(zhēng)的問(wèn)題?。。?/li>

第一種實(shí)現(xiàn)：

# -*- coding: utf-8 -*-
from threading import Thread, Lock

a = 0
n = 100000   # 指定加減次數(shù)
# 線程鎖
lock = Lock()
def incr(n):
    global  a
    # 對(duì)全局變量a做n次加1
    for i in range(n):
        lock.acquire()
        a += 1
        lock.release()
def decr(n):
    global a
    # 對(duì)全局變量a做n次減一
    for i in range(n):
        lock.acquire()
        a -= 1
        lock.release()
t_incr = Thread(target=incr, args=(n, ))
t_decr = Thread(target=decr, args=(n, ))
t_incr.start(); t_decr.start()
t_incr.join();  t_decr.join()
print(a)

第二種實(shí)現(xiàn)：

# -*- coding: utf-8 -*-
from threading import Thread, Lock

a = 0
n = 100000   # 指定加減次數(shù)
# 線程鎖
lock = Lock()

def incr(n):
    global  a
    # 對(duì)全局變量a做n次加1
    for i in range(n):
        with lock:
            a += 1
def decr(n):
    global a
    # 對(duì)全局變量a做n次減一
    for i in range(n):
        with lock:
            a -= 1

t_incr = Thread(target=incr, args=(n, ))
t_decr = Thread(target=decr, args=(n, ))
t_incr.start(); t_decr.start()
t_incr.join();  t_decr.join()
print(a)

分析此階段，我們會(huì)發(fā)現(xiàn)進(jìn)程和線程的痛點(diǎn)！??！

線程之間如何進(jìn)行協(xié)作？

最典型的例子就是生產(chǎn)者/消費(fèi)者模式：若干個(gè)生產(chǎn)者線程向隊(duì)列中寫(xiě)入數(shù)據(jù)，若干個(gè)消費(fèi)者線程從隊(duì)列中消費(fèi)數(shù)據(jù)。
（功能?。?/p>

1.定義了一個(gè)生產(chǎn)者類，一個(gè)消費(fèi)者類。
2.生產(chǎn)者類循環(huán)100次，向同步隊(duì)列當(dāng)中插入數(shù)據(jù)。
3.消費(fèi)者循環(huán)監(jiān)聽(tīng)同步隊(duì)列，當(dāng)隊(duì)列有數(shù)據(jù)時(shí)拉取數(shù)據(jù)。
4.如果隊(duì)列滿了（達(dá)到5個(gè)元素），生產(chǎn)者阻塞。
5.如果隊(duì)列空了，消費(fèi)者阻塞。

這里就引入了協(xié)程！是一種比線程更加輕量級(jí)的存在。正如一個(gè)進(jìn)程可以擁有多個(gè)線程一樣，一個(gè)線程也可以擁有多個(gè)協(xié)程。
最重要的是，協(xié)程不是被操作系統(tǒng)內(nèi)核所管理，而完全是由程序所控制（也就是在用戶態(tài)執(zhí)行）。這樣帶來(lái)的好處就是性能得到了很大的提升，不會(huì)像線程切換那樣消耗資源。

代碼走起來(lái)（依舊是生產(chǎn)者/消費(fèi)者模式的例子?。?

def consumer():
? ? while True:
? ? ? ? # consumer協(xié)程等待接收數(shù)據(jù)
? ? ? ? number = yield
? ? ? ? print('開(kāi)始消費(fèi)', number)

consumer_result = consumer()
# 讓初始化狀態(tài)的consumer協(xié)程先執(zhí)行起來(lái)，在yield處停止
next(consumer_result)

for num in range(100):
? ? print('開(kāi)始生產(chǎn)', num)
? ? # 發(fā)送數(shù)據(jù)給consumer協(xié)程
? ? consumer_result.send(num)

代碼中創(chuàng)建了一個(gè)叫做consumer_result的協(xié)程，并且在主線程中生產(chǎn)數(shù)據(jù)，協(xié)程中消費(fèi)數(shù)據(jù)。
其中 yield 是python當(dāng)中的語(yǔ)法。當(dāng)協(xié)程執(zhí)行到y(tǒng)ield關(guān)鍵字時(shí)，會(huì)暫停在那一行，等到主線程調(diào)用send方法發(fā)送了數(shù)據(jù)，協(xié)程才會(huì)接到數(shù)據(jù)繼續(xù)執(zhí)行。
但是，yield讓協(xié)程暫停，和線程的阻塞是有本質(zhì)區(qū)別的。協(xié)程的暫停完全由程序控制，線程的阻塞狀態(tài)是由操作系統(tǒng)內(nèi)核來(lái)進(jìn)行切換。

因此，協(xié)程的開(kāi)銷遠(yuǎn)遠(yuǎn)小于線程的開(kāi)銷!!!

執(zhí)行結(jié)果：

2. 隊(duì)列的基本概念

一個(gè)入口，一個(gè)出口；
先入先出（FIFO）。

import queue

隊(duì)列操作一覽:

入隊(duì)： put(item)
出隊(duì)： get()
測(cè)試空： empty()
測(cè)試滿： full()
隊(duì)列長(zhǎng)度： qsize()
任務(wù)結(jié)束： task_done()
等待完成： join()

注意：

get()等待任務(wù)完成，如果不加task_done()則不表示任務(wù)完成，只要加這句才表明完成。才會(huì)結(jié)束執(zhí)行。
join就是阻塞，直到這個(gè)任務(wù)完成（完成的標(biāo)準(zhǔn)就是每次取出都task_done()了）

簡(jiǎn)單使用隊(duì)列的方法：

# -*- coding: utf-8 -*-
import queue

# 創(chuàng)建隊(duì)列
q = queue.Queue(4)

# 入隊(duì)
q.put(1)
q.put(2)
q.put(3)
print(q.full())
q.put(4)
print(q.full())

# 出隊(duì)
print(q.get())
print(q.get())
print(q.empty())
print(q.get())
print(q.get())
print(q.empty())