Python 并发编程中的调试技巧：快速找出程序中的错误

import threadingdef task(num):print("Thread {} is running".fORMat(num))def main():threads = []for i in range(5):t = threading.Thread(target=task, args=(i,))threads.append(t)for t in threads:t.start()for t in threads:t.join()if __name__ == "__main__":main()

import pdb# 在要调试的代码行设置断点pdb.set_trace()

import threadingimport timeclass Counter:def __init__(self):self.count = 0self.lock = threading.Lock()def increment(self):with self.lock:self.count += 1def main():counter = Counter()threads = []for i in range(100):t = threading.Thread(target=counter.increment)threads.append(t)for t in threads:t.start()for t in threads:t.join()print(counter.count)if __name__ == "__main__":main()

import multiprocessing as mpdef task(queue):data = queue.get()print("Process {} received data: {}".format(mp.current_process().pid, data))def main():queue = mp.Queue()processes = []for i in range(5):p = mp.Process(target=task, args=(queue,))processes.append(p)for p in processes:p.start()for p in processes:queue.put(i)for p in processes:p.join()if __name__ == "__main__":main()

import threadingimport timedef task(num):if num % 2 == 0:raise ValueError("Even number: {}".format(num))else:print("Thread {} is running".format(num))def main():threads = []for i in range(5):t = threading.Thread(target=task, args=(i,))threads.append(t)for t in threads:t.start()for t in threads:t.join()if __name__ == "__main__":main()