python并發(fā)編程多進(jìn)程模擬搶票實(shí)現(xiàn)過程

更新時(shí)間：2019年08月20日 15:10:21 作者：minger_lcm

這篇文章主要介紹了python并發(fā)編程多進(jìn)程模擬搶票實(shí)現(xiàn)過程,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

搶票是并發(fā)執(zhí)行

多個(gè)進(jìn)程可以訪問同一個(gè)文件

多個(gè)進(jìn)程共享同一文件，我們可以把文件當(dāng)數(shù)據(jù)庫(kù)，用多個(gè)進(jìn)程模擬多個(gè)人執(zhí)行搶票任務(wù)

db.txt

{"count": 1}

并發(fā)運(yùn)行，效率高，但競(jìng)爭(zhēng)寫同一文件，數(shù)據(jù)寫入錯(cuò)亂,只有一張票，都賣成功給了10個(gè)人

#文件db.txt的內(nèi)容為：{"count":1}
#注意一定要用雙引號(hào)，不然json無法識(shí)別
from multiprocessing import Process
import time
import json
class Foo(object):
  def search(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)
      time.sleep(1) # 模擬讀數(shù)據(jù)的網(wǎng)絡(luò)延遲
      print("<%s>用戶 查看剩余票數(shù)為 [%s]" % (name, dic["count"]))
  def get(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)
      if dic["count"] > 0:
        dic["count"] -= 1
        time.sleep(1) # 模擬寫數(shù)據(jù)的網(wǎng)絡(luò)延遲
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 購(gòu)票成功" % name)
          print("剩余票數(shù)為 [%s]" % dic["count"])
      else:
        print("沒票了，搶光了")
  def task(self, name):
    self.search(name)
    self.get(name)
if __name__ == "__main__":
  obj = Foo()
  for i in range(1,11):  # 模擬并發(fā)10個(gè)客戶端搶票
    p = Process(target=obj.task, args=("路人%s" % i,))
    p.start()

總結(jié)：程序出現(xiàn)數(shù)據(jù)寫入錯(cuò)亂

大家都查到票為1，都購(gòu)票成功

<路人1>用戶 查看剩余票數(shù)為 [1]
<路人2>用戶 查看剩余票數(shù)為 [1]
<路人3>用戶 查看剩余票數(shù)為 [1]
<路人4>用戶 查看剩余票數(shù)為 [1]
<路人5>用戶 查看剩余票數(shù)為 [1]
<路人6>用戶 查看剩余票數(shù)為 [1]
<路人7>用戶 查看剩余票數(shù)為 [1]
<路人8>用戶 查看剩余票數(shù)為 [1]
<路人9>用戶 查看剩余票數(shù)為 [1]
<路人10>用戶 查看剩余票數(shù)為 [1]
<路人1> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人2> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人3> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人4> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人5> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人6> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人7> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人8> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人9> 購(gòu)票成功
剩余票數(shù)為 [0]
<路人10> 購(gòu)票成功
剩余票數(shù)為 [0]

總結(jié)程序出現(xiàn)數(shù)據(jù)寫入錯(cuò)亂

加鎖處理：購(gòu)票行為由并發(fā)變成了串行，犧牲了運(yùn)行效率，但保證了數(shù)據(jù)安全

購(gòu)票功能不應(yīng)該并發(fā)執(zhí)行，查票應(yīng)該是并發(fā)執(zhí)行的

查票準(zhǔn)不準(zhǔn)確不重要，有可能這張票就被別人買走

一個(gè)人寫完以后，讓另外一個(gè)人基于上一個(gè)人寫的結(jié)果，再做購(gòu)票操作

#把文件db.txt的內(nèi)容重置為：{"count":1}
from multiprocessing import Process
from multiprocessing import Lock
import time
import json
class Foo(object):
  def search(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)
      time.sleep(1) # 模擬讀數(shù)據(jù)的網(wǎng)絡(luò)延遲
      print("<%s>用戶 查看剩余票數(shù)為 [%s]" % (name, dic["count"]))
  def get(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)
      if dic["count"] > 0:
        dic["count"] -= 1
        time.sleep(1) # 模擬寫數(shù)據(jù)的網(wǎng)絡(luò)延遲
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 購(gòu)票成功" % name)
          print("剩余票數(shù)為 [%s]" % dic["count"])
      else:
        print("沒票了，搶光了")
  def task(self, name, mutex):
    self.search(name)
    mutex.acquire()
    self.get(name)
    mutex.release()
if __name__ == "__main__":
  mutex = Lock()
  obj = Foo()
  for i in range(1,11): # 模擬并發(fā)10個(gè)客戶端搶票
    p = Process(target=obj.task, args=("路人%s" % i, mutex))
    p.start()

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

<路人2>用戶 查看剩余票數(shù)為 [1]
<路人3>用戶 查看剩余票數(shù)為 [1]
<路人1>用戶 查看剩余票數(shù)為 [1]
<路人4>用戶 查看剩余票數(shù)為 [1]
<路人5>用戶 查看剩余票數(shù)為 [1]
<路人7>用戶 查看剩余票數(shù)為 [1]
<路人6>用戶 查看剩余票數(shù)為 [1]
<路人8>用戶 查看剩余票數(shù)為 [1]
<路人9>用戶 查看剩余票數(shù)為 [1]
<路人10>用戶 查看剩余票數(shù)為 [1]
<路人2> 購(gòu)票成功
剩余票數(shù)為 [0]
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了
沒票了，搶光了

with lock

相當(dāng)于lock.acquire(),執(zhí)行完自代碼塊自動(dòng)執(zhí)行l(wèi)ock.release()

from multiprocessing import Process
from multiprocessing import Lock
import time
import json
class Foo(object):
  def search(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)

      time.sleep(1) # 模擬讀數(shù)據(jù)的網(wǎng)絡(luò)延遲
      print("<%s>用戶 查看剩余票數(shù)為 [%s]" % (name, dic["count"]))
  def get(self, name):
    with open("db.txt", "r") as f_read:
      dic = json.load(f_read)
      if dic["count"] > 0:
        dic["count"] -= 1
        time.sleep(1) # 模擬寫數(shù)據(jù)的網(wǎng)絡(luò)延遲
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 購(gòu)票成功" % name)
          print("剩余票數(shù)為 [%s]" % dic["count"])
      else:
        print("沒票了，搶光了")
  def task(self, name, mutex):
    self.search(name)
    with mutex: # 相當(dāng)于lock.acquire(),執(zhí)行完自代碼塊自動(dòng)執(zhí)行l(wèi)ock.release()
      self.get(name)
if __name__ == "__main__":
  mutex = Lock()
  obj = Foo()
  for i in range(1,11): # 模擬并發(fā)10個(gè)客戶端搶票
    p = Process(target=obj.task, args=("路人%s" % i, mutex))
    p.start()

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: