Python的執行緒使用佇列Queue來改造轉賬場景

2022-02-24 13:00:40

一、看看轉賬場景的問題

前面有兩篇文章展示了轉賬反覆讀寫amount，導致結果出錯。

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

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

我們前幾篇使用多個執行緒反覆轉長：+1和-1。

按常理，結果應該仍舊是100.

這個是全部程式碼：

import random
import threading
import datetime
import time

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


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


# 建立20個任務重複給學委賬戶轉賬
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("活躍執行緒數:", threading.active_count())
print("活躍執行緒:", threading.current_thread().name)
print("學委賬戶餘額：", xuewei_account)

等待所有轉賬執行緒執行結束，我們看到結果是錯誤的：

二、這種問題怎麼使用佇列來解決呢？

前面說了，多執行緒反覆讀寫共用資料，是問題的根源。

改程式碼為同步互斥模式，保證任意一個時間一個執行緒更新共用資料，那麼問題就解決了。（這前面也展示了，用的是Lock鎖的方案）

這個能怎麼用佇列呢？

可以先思考10秒，根據學習到的加鎖和佇列的特性，想想這個怎麼做。

好，答案現在揭曉：

Queue這個佇列有多個函數，一個是put函數，一個是get函數。

一個負責放入資料到隊尾，一個可以從對頭取出元素。

剛好適合轉賬業務，我們是不是可以把每次轉賬操作變成一個一個指令/事件。 比如下面的：

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個執行緒，每個10萬次資料讀寫，共200萬個事件。

所以我們可以把這個事情轉換為：200萬個轉賬事件。

因為Queue是執行緒安全的，所以我們可以並行200萬次轉賬，另外交給一執行緒進行轉賬處理。

這樣就保證每次只有一個執行緒對xuewei_account學委賬戶進行讀寫。

改造，使用佇列來解決問題

展示程式碼：

import random
import threading
import datetime
import time
import queue

q = queue.Queue()

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


# amount為負數即是轉出金額
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監控", target=monitor_q)
q_thread.start()
# 建立20個任務重複給學委賬戶轉賬
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="處理轉賬專線", target=handle_amount)
vip_thread.start()

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

print("-" * 16)
print("活躍執行緒數:", threading.active_count())
print("活躍執行緒:", threading.current_thread().name)
print("學委賬戶餘額：", xuewei_account)

這裡執行了多個執行緒執行轉賬（傳送轉賬金額進佇列）。

然後執行一個vip通道（單獨執行緒）處理學委賬戶的轉賬業務。

同時也執行了一個監控佇列的執行緒，每隔一段時間列印佇列的任務情況。

下面是執行結果，執行幾次結果都是正確的。