Python多进程并发抢票实战：从数据错乱到加锁优化

江岸

发布时间：2026-10-13 | 2人浏览过

原创

本文通过Python多进程模拟高并发抢票场景，演示了未加锁时因读写竞争导致的数据错乱问题。随后引入互斥锁（Lock）机制，将并发购票串行化，确保数据一致性。文章对比了基础实现与加锁后的执行结果，并介绍了with语句在锁管理中的简化用法，帮助开发者理解并发安全的重要性。

多进程并发抢票基础实现

在并发编程中，抢票是一个典型的并发执行场景。

多个进程能够同时访问同一个文件资源。

通过让多个进程共享同一文件，我们可以将该文件视为数据库，进而利用多个进程来模拟多用户同时执行抢票任务的过程。

数据文件命名为db.txt。


{"count": 1}

虽然并发运行能显著提升效率，但由于多个进程竞争写入同一文件，会导致数据写入错乱。例如，系统中仅有一张票，却可能因为竞态条件被错误地卖给10个人。


#文件db.txt的内容为：{"count":1}
#注意一定要用双引号，不然json无法识别
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) # 模拟读数据的网络延迟
      print("<%s>用户 查看剩余票数为 [%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) # 模拟写数据的网络延迟
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 购票成功" % name)
          print("剩余票数为 [%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):  # 模拟并发10个客户端抢票
    p = Process(target=obj.task, args=("路人%s" % i,))
    p.start()

数据竞争问题分析

总结来看，程序出现了数据写入错乱的现象。

所有用户查询时均显示票数为1，且最终都显示购票成功。


<路人1>用户 查看剩余票数为 [1]
<路人2>用户 查看剩余票数为 [1]
<路人3>用户 查看剩余票数为 [1]
<路人4>用户 查看剩余票数为 [1]
<路人5>用户 查看剩余票数为 [1]
<路人6>用户 查看剩余票数为 [1]
<路人7>用户 查看剩余票数为 [1]
<路人8>用户 查看剩余票数为 [1]
<路人9>用户 查看剩余票数为 [1]
<路人10>用户 查看剩余票数为 [1]
<路人1> 购票成功
剩余票数为 [0]
<路人2> 购票成功
剩余票数为 [0]
<路人3> 购票成功
剩余票数为 [0]
<路人4> 购票成功
剩余票数为 [0]
<路人5> 购票成功
剩余票数为 [0]
<路人6> 购票成功
剩余票数为 [0]
<路人7> 购票成功
剩余票数为 [0]
<路人8> 购票成功
剩余票数为 [0]
<路人9> 购票成功
剩余票数为 [0]
<路人10> 购票成功
剩余票数为 [0]

总结程序出现数据写入错乱

互斥锁解决方案与优化

为解决此问题，需引入加锁处理：将购票行为由并发执行改为串行执行。虽然这牺牲了部分运行效率，但有效保证了数据的安全性。

从逻辑上讲，购票操作本身不应并发执行，而查票环节则可以是并发的。

需要注意的是，查票结果的准确性并不绝对重要，因为有可能在查询到票之后、提交之前，这张票已被其他用户买走。

正确的逻辑是：一个人完成写入后，其他人才基于该最新结果进行后续的购票操作。


#把文件db.txt的内容重置为：{"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) # 模拟读数据的网络延迟
      print("<%s>用户 查看剩余票数为 [%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) # 模拟写数据的网络延迟
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 购票成功" % name)
          print("剩余票数为 [%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): # 模拟并发10个客户端抢票
    p = Process(target=obj.task, args=("路人%s" % i, mutex))
    p.start()

以下是具体的执行结果展示。


<路人2>用户 查看剩余票数为 [1]
<路人3>用户 查看剩余票数为 [1]
<路人1>用户 查看剩余票数为 [1]
<路人4>用户 查看剩余票数为 [1]
<路人5>用户 查看剩余票数为 [1]
<路人7>用户 查看剩余票数为 [1]
<路人6>用户 查看剩余票数为 [1]
<路人8>用户 查看剩余票数为 [1]
<路人9>用户 查看剩余票数为 [1]
<路人10>用户 查看剩余票数为 [1]
<路人2> 购票成功
剩余票数为 [0]
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了
没票了，抢光了

使用with语句配合锁对象。

这相当于先执行lock.acquire()，在代码块执行完毕后自动执行lock.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) # 模拟读数据的网络延迟
      print("<%s>用户 查看剩余票数为 [%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) # 模拟写数据的网络延迟
        with open("db.txt", "w") as f_write:
          json.dump(dic, f_write)
          print("<%s> 购票成功" % name)
          print("剩余票数为 [%s]" % dic["count"])
      else:
        print("没票了，抢光了")
  def task(self, name, mutex):
    self.search(name)
    with mutex: # 相当于lock.acquire(),执行完自代码块自动执行lock.release()
      self.get(name)
if __name__ == "__main__":
  mutex = Lock()
  obj = Foo()
  for i in range(1,11): # 模拟并发10个客户端抢票
    p = Process(target=obj.task, args=("路人%s" % i, mutex))
    p.start()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。