Python的線程使用隊(duì)列Queue來改造轉(zhuǎn)賬場景

更新時(shí)間：2022年02月24日 11:27:56 作者：雷學(xué)委

前篇我們了隊(duì)列Queue和轉(zhuǎn)賬場景這次趁熱學(xué)委展示一下使用隊(duì)列解決轉(zhuǎn)賬場景的問題，這篇文章主要介紹了Python的線程使用隊(duì)列來改造轉(zhuǎn)賬場景,需要的朋友可以參考一下

一、看看轉(zhuǎn)賬場景的問題

前面有兩篇文章展示了轉(zhuǎn)賬反復(fù)讀寫amount，導(dǎo)致結(jié)果出錯(cuò)。

xuewei_account = dict()
xuewei_account['amount'] = 100

# amount為負(fù)數(shù)即是轉(zhuǎn)出金額
def transfer(money):
? ? for i in range(100000):
? ? ? ? xuewei_account['amount'] = xuewei_account['amount'] + money

我們前幾篇使用多個(gè)線程反復(fù)轉(zhuǎn)長：+1和-1。

按常理，結(jié)果應(yīng)該仍舊是100.

這個(gè)是全部代碼：

import random
import threading
import datetime
import time

xuewei_account = dict()
xuewei_account['amount'] = 100


# amount為負(fù)數(shù)即是轉(zhuǎn)出金額
def transfer(money):
? ? for i in range(100000):
? ? ? ? xuewei_account['amount'] = xuewei_account['amount'] + money


# 創(chuàng)建20個(gè)任務(wù)重復(fù)給學(xué)委賬戶轉(zhuǎn)賬
threads = []
for i in range(10):
? ? t1 = threading.Thread(target=lambda: transfer(-1))
? ? threads.append(t1)
? ? t2 = threading.Thread(target=lambda: transfer(1))
? ? threads.append(t2)

for t in threads:
? ? t.start()
for t in threads:
? ? t.join()

print("-" * 16)
print("活躍線程數(shù):", threading.active_count())
print("活躍線程:", threading.current_thread().name)
print("學(xué)委賬戶余額：", xuewei_account)

等待所有轉(zhuǎn)賬線程運(yùn)行結(jié)束，我們看到結(jié)果是錯(cuò)誤的：

二、這種問題怎么使用隊(duì)列來解決呢？

前面說了，多線程反復(fù)讀寫共享數(shù)據(jù)，是問題的根源。

改代碼為同步互斥模式，保證任意一個(gè)時(shí)間一個(gè)線程更新共享數(shù)據(jù)，那么問題就解決了。（這前面也展示了，用的是Lock鎖的方案）

這個(gè)能怎么用隊(duì)列呢？

可以先思考10秒，根據(jù)學(xué)習(xí)到的加鎖和隊(duì)列的特性，想想這個(gè)怎么做。

好，答案現(xiàn)在揭曉：

Queue這個(gè)隊(duì)列有多個(gè)函數(shù)，一個(gè)是put函數(shù)，一個(gè)是get函數(shù)。

一個(gè)負(fù)責(zé)放入數(shù)據(jù)到隊(duì)尾，一個(gè)可以從對頭取出元素。

剛好適合轉(zhuǎn)賬業(yè)務(wù)，我們是不是可以把每次轉(zhuǎn)賬操作變成一個(gè)一個(gè)指令/事件。 比如下面的：

event(amount=1,acount=xuewei_account)
....
event(amount=-1,acount=xuewei_account)
....
event(amount=1,acount=xuewei_account)
....
event(amount=-1,acount=xuewei_account)

20個(gè)線程，每個(gè)10萬次數(shù)據(jù)讀寫，共200萬個(gè)事件。

所以我們可以把這個(gè)事情轉(zhuǎn)換為：200萬個(gè)轉(zhuǎn)賬事件。

因?yàn)镼ueue是線程安全的，所以我們可以并發(fā)200萬次轉(zhuǎn)賬，另外交給一線程進(jìn)行轉(zhuǎn)賬處理。

這樣就保證每次只有一個(gè)線程對xuewei_account學(xué)委賬戶進(jìn)行讀寫。

改造，使用隊(duì)列來解決問題

展示代碼：

import random
import threading
import datetime
import time
import queue

q = queue.Queue()

xuewei_account = dict()
xuewei_account['amount'] = 100


# amount為負(fù)數(shù)即是轉(zhuǎn)出金額
def transfer(money):
? ? for i in range(100000):
? ? ? ? q.put(money)


def handle_amount():
? ? while not q.empty():
? ? ? ? amount = q.get()
? ? ? ? xuewei_account['amount'] += amount


def monitor_q():
? ? counter = 0
? ? time.sleep(3)
? ? while counter < 1000 and not q.empty():
? ? ? ? print("q size:", q.qsize())
? ? ? ? time.sleep(3)
? ? ? ? counter+=1


q_thread = threading.Thread(name="Q監(jiān)控", target=monitor_q)
q_thread.start()
# 創(chuàng)建20個(gè)任務(wù)重復(fù)給學(xué)委賬戶轉(zhuǎn)賬
threads = []
for i in range(10):
? ? t1 = threading.Thread(target=lambda: transfer(-1))
? ? threads.append(t1)
? ? t2 = threading.Thread(target=lambda: transfer(1))
? ? threads.append(t2)

for t in threads:
? ? t.start()

vip_thread = threading.Thread(name="處理轉(zhuǎn)賬專線", target=handle_amount)
vip_thread.start()

for t in threads:
? ? t.join()
vip_thread.join()

print("-" * 16)
print("活躍線程數(shù):", threading.active_count())
print("活躍線程:", threading.current_thread().name)
print("學(xué)委賬戶余額：", xuewei_account)

這里運(yùn)行了多個(gè)線程執(zhí)行轉(zhuǎn)賬（發(fā)送轉(zhuǎn)賬金額進(jìn)隊(duì)列）。

然后運(yùn)行一個(gè)vip通道（單獨(dú)線程）處理學(xué)委賬戶的轉(zhuǎn)賬業(yè)務(wù)。

同時(shí)也運(yùn)行了一個(gè)監(jiān)控隊(duì)列的線程，每隔一段時(shí)間打印隊(duì)列的任務(wù)情況。

下面是運(yùn)行結(jié)果，運(yùn)行幾次結(jié)果都是正確的。