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

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

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

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

這個是全部代碼：

import random
import threading
import datetime
import time

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


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


# 創(chuàng)建20個任務(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)賬線程運行結(jié)束，我們看到結(jié)果是錯誤的：

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

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

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

這個能怎么用隊列呢？

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

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

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

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

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

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萬次數(shù)據(jù)讀寫，共200萬個事件。

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

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

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

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

展示代碼：

import random
import threading
import datetime
import time
import queue

q = queue.Queue()

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


# amount為負數(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個任務(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)

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

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

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

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