#!/usr/bin/env python # -*- coding:utf-8 -*- import threading import time NUM = 0 def show(): global NUM NUM += 1 name = t.getName() time.sleep(1) # 注意，这行语句的位置很重要，必须在NUM被修改后，否则观察不到脏数据的现象。 print(name, "执行完毕后，NUM的值为： ", NUM) for i in range(10): t = threading.Thread(target=show) t.start()
import time import threading NUM = 10 def func(lock): global NUM lock.acquire() # 让锁开始起作用 NUM -= 1 time.sleep(1) print(NUM) lock.release() # 释放锁 lock = threading.Lock() # 实例化一个锁对象 for i in range(10): t = threading.Thread(target=func, args=(lock,)) # 记得把锁当作参数传递给func参数 t.start()
import threading def condition(): ret = False r = input(">>>") if r == "yes": ret = True return ret def func(conn, i): print(i) conn.acquire() conn.wait_for(condition) # 这个方法接受一个函数的返回值 print(i+100) conn.release() c = threading.Condition() for i in range(10): t = threading.Thread(target=func, args=(c, i,)) t.start()
#!/usr/bin/env python # -*- coding:utf-8 -*- import time import queue import threading q = queue.Queue(10) def productor(i): while True: q.put("厨师 %s 做的包子！"%i) time.sleep(2) def consumer(k): while True: print("顾客 %s 吃了一个 %s"%(k,q.get())) time.sleep(1) for i in range(3): t = threading.Thread(target=productor,args=(i,)) t.start() for k in range(10): v = threading.Thread(target=consumer,args=(k,)) v.start()
#!/usr/bin/env python # -*- coding:utf-8 -*- import queue import time import threading class MyThreadPool: def __init__(self, maxsize=5): self.maxsize = maxsize self._q = queue.Queue(maxsize) for i in range(maxsize): self._q.put(threading.Thread)
def get_thread(self): return self._q.get()
def add_thread(self): self._q.put(threading.Thread) def task(i, pool): print(i) time.sleep(1) pool.add_thread() pool = MyThreadPool(5) for i in range(100): t = pool.get_thread() obj = t(target=task, args=(i,pool)) obj.start()
self.q.empty() @contextlib.contextmanager def worker_state(self, state_list, worker_thread): """ 用于记录线程中正在等待的线程数 """ state_list.append(worker_thread) try: yield finally: state_list.remove(worker_thread) # How to use pool = ThreadPool(5) def callback(status, result): # status, execute action status # # result, execute action return value # pass def action(i): print(i) for i in range(30): ret = pool.run(action, (i,), callback) time.sleep(5) print(len(pool.generate_list), len(pool.free_list)) print(len(pool.generate_list), len(pool.free_list)) # pool.close() # pool.terminate()
二、进程
在python中multiprocess模块提供了Process类，实现进程相关的功能。但是，由于它是基于fork机制的，因此不被windows平台支持。想要在windows中运行，必须使用if name == ‘main:的方式，显然这只能用于调试和学习，不能用于实际环境。 （PS：在这里我必须吐槽一下python的包、模块和类的组织结构。在multiprocess中你既可以import大写的Process，也可以import小写的process，这两者是完全不同的东西。这种情况在python中很多，新手容易傻傻分不清。） 下面是一个简单的多进程例子，你会发现Process的用法和Thread的用法几乎一模一样。
1 2 3 4 5 6 7 8 9 10
from multiprocessing import Process def foo(i): print("This is Process ", i) if __name__ == '__main__': for i in range(5): p = Process(target=foo, args=(i,)) p.start()
2.1 进程的数据共享
每个进程都有自己独立的数据空间，不同进程之间通常是不能共享数据，创建一个进程需要非常大的开销。
1 2 3 4 5 6 7 8 9 10 11 12 13
from multiprocessing import Process list_1 = [] def foo(i): list_1.append(i) print("This is Process ", i," and list_1 is ", list_1) if __name__ == '__main__': for i in range(5): p = Process(target=foo, args=(i,)) p.start() print("The end of list_1:", list_1)
from multiprocessing import Process from multiprocessing import Array def Foo(i,temp): temp[0] += 100 for item in temp: print(i,'----->',item) if __name__ == '__main__': temp = Array('i', [11, 22, 33, 44]) for i in range(2): p = Process(target=Foo, args=(i,temp)) p.start()
from multiprocessing import Process,Manager def Foo(i,dic): dic[i] = 100+i print(dic.values()) if __name__ == '__main__': manage = Manager() dic = manage.dict() for i in range(10): p = Process(target=Foo, args=(i,dic)) p.start() p.join()
Manager比Array要好用一点，因为它可以同时保存多种类型的数据格式。
2.1.3 使用queues的Queue类共享数据
1 2 3 4 5 6 7 8 9 10 11 12 13 14
import multiprocessing from multiprocessing import Process from multiprocessing import queues def foo(i,arg): arg.put(i) print('The Process is ', i, "and the queue's size is ", arg.qsize()) if __name__ == "__main__": li = queues.Queue(20, ctx=multiprocessing) for i in range(10): p = Process(target=foo, args=(i,li,)) p.start()
from multiprocessing import Process from multiprocessing import queues from multiprocessing import Array from multiprocessing import RLock, Lock, Event, Condition, Semaphore import multiprocessing import time def foo(i,lis,lc): lc.acquire() lis[0] = lis[0] - 1 time.sleep(1) print('say hi',lis[0]) lc.release() if __name__ == "__main__": # li = [] li = Array('i', 1) li[0] = 10 lock = RLock() for i in range(10): p = Process(target=foo,args=(i,li,lock)) p.start()