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

多個(gè)進(jìn)程可以訪問(wè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)寫(xiě)同一文件，數(shù)據(jù)寫(xiě)入錯(cuò)亂,只有一張票，都賣(mài)成功給了10個(gè)人

#文件db.txt的內(nèi)容為：{"count":1}

#注意一定要用雙引號(hào)，不然json無(wú)法識(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("用戶(hù) 查看剩余票數(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) # 模擬寫(xiě)數(shù)據(jù)的網(wǎng)絡(luò)延遲

with open("db.txt", "w") as f_write:

json.dump(dic, f_write)

print(" 購(gòu)票成功" % name)

print("剩余票數(shù)為 [%s]" % dic["count"])

else:

print("沒(méi)票了，搶光了")

def task(self, name):

self.search(name)

self.get(name)

if __name__ == "__main__":

obj = Foo()

for i in range(1,11): # 模擬并發(fā)10個(gè)客戶(hù)端搶票

p = Process(target=obj.task, args=("路人%s" % i,))

p.start()

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

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

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

購(gòu)票成功

剩余票數(shù)為 [0]

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

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

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

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

一個(gè)人寫(xiě)完以后，讓另外一個(gè)人基于上一個(gè)人寫(xiě)的結(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("用戶(hù) 查看剩余票數(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) # 模擬寫(xiě)數(shù)據(jù)的網(wǎng)絡(luò)延遲

with open("db.txt", "w") as f_write:

json.dump(dic, f_write)

print(" 購(gòu)票成功" % name)

print("剩余票數(shù)為 [%s]" % dic["count"])

else:

print("沒(méi)票了，搶光了")

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è)客戶(hù)端搶票

p = Process(target=obj.task, args=("路人%s" % i, mutex))

p.start()

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

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

用戶(hù) 查看剩余票數(shù)為 [1]

購(gòu)票成功

剩余票數(shù)為 [0]

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

沒(méi)票了，搶光了

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("用戶(hù) 查看剩余票數(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) # 模擬寫(xiě)數(shù)據(jù)的網(wǎng)絡(luò)延遲

with open("db.txt", "w") as f_write:

json.dump(dic, f_write)

print(" 購(gòu)票成功" % name)

print("剩余票數(shù)為 [%s]" % dic["count"])

else:

print("沒(méi)票了，搶光了")

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è)客戶(hù)端搶票

p = Process(target=obj.task, args=("路人%s" % i, mutex))

p.start()

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

本文標(biāo)題: python并發(fā)編程多進(jìn)程 模擬搶票實(shí)現(xiàn)過(guò)程

本文地址: http://www.cppcns.com/jiaoben/python/268996.html

總結(jié)

以上是生活随笔為你收集整理的python模拟app抢票_python并发编程多进程 模拟抢票实现过程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。