• def job(name): print('子线程{}--{}'.format(name,threading.current_thread().getName())) time.sleep(1) print("do something1 %s" % (name))def job1(name): print('子线程{}--{}'.format(name, threading.current_thread().getName())) time.sleep(2) print("do something2 %s" % (name))if __name__ == "__main__": t1 = threading.Thread(target=job, args=('job',)) t2 = threading.Thread(target=job1, args=('job1',)) # t2.setDaemon(True) #守护线程 t1.start() t2.start() # t1.join() # t2.join() print('主线程{}'.format(os.getpid()))

• class myThread(threading.Thread): def __init__(self, ti): threading.Thread.__init__(self) self.ti = ti def run(self): for i in range(self.ti): msg = 'I am' + self.name + '@' + str(i) time.sleep(1) print(msg)if __name__ == "__main__": t = myThread(3) t.start() t.join()print('主线程')

class SafeQueue(threading.Thread):    # 退出队列的信号    SIG_QUIT = 'sig_quit'    def __init__(self,recv_calback):        threading.Thread.__init__(self)        ## 构造线程安全队列        self.Q = queue.Queue()        self.recv_calback =recv_calback        self.start()    #放入队列    def put(self,datas):        threadName =threading.currentThread().name        # print(threadName)        self.Q.put(datas)    #关闭队列    def close(self):        self.put(SafeQueue.SIG_QUIT)    ##主循环,处理队列接收    def run(self):        while True:           try:               print('等待中消息队列任务……')               datas = self.Q.get()               if datas == SafeQueue.SIG_QUIT: #收到退出队列信号                   break                #回调客户端               self.recv_calback(datas)           except: # 抛出打断异常               break##队列回调函数def queue_callback(datas):    print("接收到数据:",datas)    ## 将子任务结果加入 全局集合    try:        array_mutex.acquire()#锁定        datas_array.append(datas)        if len(datas_array) == 4:            safeQueue.close()            print("=======大任务计算结束===========  result:", datas_array)    finally:        array_mutex.release()#释放## 子任务计算函数def calclulate(num):    threadName =threading.currentThread().name    print(threadName , ' 正在计算')    time.sleep(2)    print(threadName, ' 计算完成,加入队列')    #将结果放入队列    safeQueue.put(threadName+"' result" + str(num))#### ----------  main start   ----------#创建锁array_mutex = threading.Lock()## 存储 子任务计算结果的 集合datas_array = []##构造安全队列safeQueue = SafeQueue(queue_callback) #开辟子线程队列，在get处阻塞# safeQueue.join()for i in range(1,5):    threading.Thread(target=calclulate, args=(i,)).start()

def wait_for_event(e):    print('event通信开始等待……')    e.wait()    print('event通信时标志位是{}'.format(e.is_set()))def wait_for_event_timeout(e,t):    print('event超时通信开始等待……')    e.wait(t)    print('event超时通信时标志位是{}'.format(e.is_set()))if __name__ == '__main__':    e = threading.Event()    w1 = threading.Thread(name='block', target=wait_for_event,args=(e,))    w1.start()    w2 = threading.Thread(name='non-block', target=wait_for_event_timeout,                                 args=(e, 2))    w2.start()    time.sleep(3)  #等待3秒后超时通信在2秒时就会往下继续执行，此时标志位还未开始设置仍然时false，3s后设置标志位w1是True    e.set()    print('标志位已经被设置为True')

# 信号量sema = Semaphore(3)  # 限制同时能访问资源的数量为3，设置为1时可以防止资源竞争num = 0def foo(tid):    sema.acquire()    # with sema:    global num    for i in range(1000000):        num +=1    print('{} acquire sema'.format(tid))    print(num)    time.sleep(1)    sema.release()    print('{} release sema'.format(tid))threads = []for i in range(5):    t = Thread(target=foo, args=(i,))    threads.append(t)    t.start()for t in threads:t.join()

con = threading.Condition()num = 0# 生产者class Producer(threading.Thread):    def __init__(self):        threading.Thread.__init__(self)    def run(self):        # 锁定线程        global num        con.acquire()        while True:            print ("{}开始添加！！！".format(threading.current_thread().getName()))            num += 1            print ("火锅里面鱼丸个数：%s" % str(num))            time.sleep(1)            if num >= 5:                print ("火锅里面里面鱼丸数量已经到达5个，无法添加了！")                # 唤醒等待的线程                con.notify()  # 唤醒小伙伴开吃啦                # 等待通知                con.wait()  #调用这个方法将使线程进入Condition的等待池等待通知，并释放锁        # 释放锁        # con.release()# 消费者class Consumers(threading.Thread):    def __init__(self):        threading.Thread.__init__(self)    def run(self):        print('等待中……')        con.acquire()        print(45555)        global num        while True:            print ("{}开始吃啦！！！".format(threading.current_thread().getName()))            num -= 1            print ("火锅里面剩余鱼丸数量：%s" %str(num))            time.sleep(2)            if num <= 0:                print ("锅底没货了，赶紧加鱼丸吧！")                con.notify()  # 唤醒其它线程                # 等待通知                con.wait()        # con.release()p = Producer()p.start()c = Consumers()c.start()

threading.RLock和threading.Lock 的区别:lock = threading.Lock() #Lock对象lock.acquire()lock.acquire() #产生了死琐。lock.release()lock.release()rLock = threading.RLock() #RLock对象rLock.acquire()rLock.acquire() #在同一线程内，程序不会堵塞。rLock.release()rLock.release()

globals_num=0deffunc1(lock):    global globals_numlock.acquire()#获得锁foriinrange(1000000):try:globals_num=globals_num+1except:passlock.release()#释放锁    print('线程1结果{}'.format(globals_num))def func2(lock):globalglobals_numwithlock:foriinrange(1000000):globals_num=globals_num+1print('线程2结果{}'.format(globals_num))if __name__ == '__main__':lock=threading.Lock()#lock=threading.Semaphore(1)t1=threading.Thread(target=func1,args=(lock,))    t2 = threading.Thread(target=func2, args=(lock,))t1.start()    t2.start()'''结果：线程1结果1000000线程2结果2000000'''

#对多把锁按锁id进行排序上锁_local = threading.local()   #每个线程开辟一个独立的空间进行数据存储@contextmanagerdefacquire(flag,*locks,**kwargs):print(flag,'开始获取锁......................')    locks = sorted(locks, key=lambda x: id(x))  #按照锁ID进行排序acquired=getattr(_local,'acquired',[])#getattr()函数用于返回一个对象属性值。getattr(a, 'bar2', 3)#属性 bar2 不存在，但设置了默认值,返回3    if acquired and max(id(lock) for lock in acquired) >= id(locks[0]):        raise RuntimeError('Lock Order Violation')acquired.extend(locks)_local.acquired=acquiredtry:forlockinlocks:lock.acquire()iflock==acquired[0]:print('{}获取第一把锁{}后，执行{}里面的任务代码：'.format(flag,lock,flag))            else:print('{}获取第二把锁{}后，执行{}里面的任务代码'.format(flag,lock,flag))yieldfinally:forlockinreversed(locks):lock.release()delacquired[-len(locks):]        print('{}执行完任务逻辑后释放锁'.format(flag))x_lock=threading.Lock()y_lock = threading.Lock()g_num = 0defthread_1():globalg_numforiinrange(1):withacquire('thread1',y_lock,x_lock):print('thread1获取锁并执行完任务代码......................')            g_num += 1defthread_2():globalg_numforiinrange(1):withacquire('thread2',y_lock,x_lock):            print('thread2获取锁并执行完任务代码......................')            g_num += 1t1=threading.Thread(target=thread_1)t1.start()t2=threading.Thread(target=thread_2)t2.start()'''结果：thread1 开始获取锁......................thread1获取第一把锁<locked _thread.lock object at 0x10c34b3c8>后，执行thread1里面的任务代码：thread1获取第二把锁<locked _thread.lock object at 0x10c34b8a0>后，执行thread1里面的任务代码thread1获取锁并执行完任务代码......................thread2 开始获取锁......................thread1执行完任务逻辑后释放锁thread2获取第一把锁<locked _thread.lock object at 0x10c34b3c8>后，执行thread2里面的任务代码：thread2获取第二把锁<locked _thread.lock object at 0x10c34b8a0>后，执行thread2里面的任务代码thread2获取锁并执行完任务代码......................thread2执行完任务逻辑后释放锁'''

def sayhello (a):    print("hello: "+a)def main():    global result    seed=["a","b","c"]    start=time.time()    task_pool = threadpool.ThreadPool(3)    requests =threadpool.makeRequests(sayhello, seed)    for ret in requests:        task_pool.putRequest(ret)    task_pool.wait()    end=time.time()    time_m = end-start    print("threadpool花费时间time: "+str(time_m))    time.sleep(1)    start2=time.time()    p = ThreadPool(3)  # 直接使用内置的    for i in seed:        p.apply_async(sayhello, args=(i,))    p.close()    p.join()    end2=time.time()    print("内置线程池花费时间time2: "+str(end2-start2))if __name__ == '__main__':    main()

• ef sayhello(a): print("hello: "+a)def main(): seed=["a","b","c"] with ThreadPoolExecutor(3) as executor: executor.submit(sayhello,str(seed)) #submit time.sleep(2) with ThreadPoolExecutor(3) as executor1: executor1.map(sayhello,seed) #mapif __name__ == '__main__': main()

• #参数times用来模拟网络请求的时间defget_html(times):time.sleep(times)print("getpage{}sfinished".format(times))returntimesexecutor=ThreadPoolExecutor(max_workers=2)#通过submit函数提交执行的函数到线程池中，submit函数立即返回，不阻塞task1=executor.submit(get_html,(3))task2=executor.submit(get_html,(2))# done方法用于判定某个任务是否完成,完成返回True，未执行也返回Trueprint('任务是否完成{}'.format(task1.done()))#cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功print('取消某个任务{}'.format(task2.cancel()))time.sleep(4)print(task1.done())print(task2.done())# result方法可以获取task的执行结果print('执行结果{}'.format(task1.result()))print('='*50) """说明：ThreadPoolExecutor构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目。使用submit函数来提交线程需要执行的任务（函数名和参数）到线程池中，并返回该任务的句柄（类似于文件、画图），注意submit()不是阻塞的，而是立即返回。通过submit函数返回的任务句柄，能够使用done()方法判断该任务是否结束。上面的例子可以看出，由于任务有2s的延时，在task1提交后立刻判断，task1还未完成，而在延时4s之后判断，task1就完成了。使用cancel()方法可以取消提交的任务，如果任务已经在线程池中运行了，就取消不了。这个例子中，线程池的大小设置为2，任务已经在运行了，所以取消失败。如果改变线程池的大小为1，那么先提交的是task1，task2还在排队等候，这是时候就可以成功取消。使用result()方法可以获取任务的返回值。查看内部代码，发现这个方法是阻塞的。"""defget_html(times):time.sleep(times)print("getpage{}sfinished".format(times)) return timesexecutor=ThreadPoolExecutor(max_workers=2)urls=[3,2,4]#并不是真的urlall_task = [executor.submit(get_html, (url)) for url in urls] #每次都需要提交一个目标函数和对应的参数print('所有任务{}'.format(all_task))forfutureinas_completed(all_task):#在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务data=future.result()print("inmain:getpage{}ssuccess".format(data))print('='*50)"""说明：上面虽然提供了判断任务是否结束的方法，但是不能在主线程中一直判断。有时候我们是得知某个任务结束了，就去获取结果，而不是一直判断每个任务有没有结束。这是就可以使用as_completed方法一次取出所有任务的结果。as_completed()方法是一个生成器，在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务，就能执行for循环下面的语句，然后继续阻塞住，循环到所有的任务结束。从结果也可以看出，先完成的任务会先通知主线程."""defget_html(times):time.sleep(times)print("getpage{}sfinished".format(times))returntimesexecutor=ThreadPoolExecutor(max_workers=2)urls=[3,2,4]#并不是真的urlall_task=[executor.submit(get_html,(url))forurlinurls]wait(all_task, timeout=3, return_when=ALL_COMPLETED)print("main")print('='*50)"""说明：wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。等待条件return_when默认为ALL_COMPLETED，表明要等待所有的任务都结束。可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main。等待条件还可以设置为FIRST_COMPLETED，表示第一个任务完成就停止等待"""

• fget_html(times):time.sleep(times)print("getpage{}sfinished".format(times)) return timesexecutor=ThreadPoolExecutor(max_workers=2)urls=[3,2,4]#并不是真的urlfor data in executor.map(get_html, urls): #函数的参数放在一个迭代器（列表，字典）里就可以 print("in main: get page {}s success".format(data)) #按照提交的顺序打印 """说明：使用map方法，无需提前使用submit方法，map方法与python标准库中的map含义相同，都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls的每个元素都执行get_html函数，并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同，输出顺序和urls列表的顺序相同，就算2s的任务先执行完成，也会先打印出3s的任务先完成，再打印2s的任务完成。"""

# greenlet创建def test1(gr, g):foriinrange(100):print("---A--")gr.switch(g,gr)#切换到另一个协程执行        time.sleep(0.5)deftest2(gr,g):foriinrange(100):print("---B--")gr.switch(g,gr)time.sleep(0.5)if __name__ == '__main__':    # 创建协程gr1=greenlet(test1)    gr2 = greenlet(test2)    # 启动切换协程    gr1.switch(gr2, gr1)   # gevent创建 monkey.patch_all() # 将程序中用到的耗时操作的代码，换为gevent中自己实现的模块,time.sleep()等效与gevent.sleep()def coroutine_work(coroutine_name):    for i in range(10):        print('当前协程是{}'.format(coroutine_name))        time.sleep(1)gevent.joinall([    gevent.spawn(coroutine_work, 'work1'),    gevent.spawn(coroutine_work, 'work2'),])

evt = Event()def setter():    print('Wait for me')    gevent.sleep(2)  # 2秒后唤醒所有在evt上等待的协程    print("Ok, I'm done")    evt.set()  # 唤醒def waiter():    print("I'll wait for you")    evt.wait()  # 等待    print('Finish waiting')gevent.joinall([    gevent.spawn(setter),    gevent.spawn(waiter),    gevent.spawn(waiter),    gevent.spawn(waiter),])

evt = AsyncResult()def setter():    print('Wait for me')    gevent.sleep(2)  # 2秒后唤醒所有在evt上等待的协程    print("Ok, I'm done")    evt.set()  # 唤醒def waiter():    print("I'll wait for you")    evt.wait()  # 等待print('Finishwaiting')gevent.joinall([    gevent.spawn(setter),    gevent.spawn(waiter),    gevent.spawn(waiter),    gevent.spawn(waiter),

q = Queue()def consumer(name):    while not q.empty():        print('%s got product %s' % (name, q.get()))        gevent.sleep(0)def producer():    for i in range(1, 4):        q.put(i)gevent.joinall([    gevent.spawn(producer),    gevent.spawn(consumer, 'steve'),    gevent.spawn(consumer, 'john'),    gevent.spawn(consumer, 'nancy'),])

• sem = BoundedSemaphore(2)def worker(n): sem.acquire() print('Worker %i acquired semaphore' % n) gevent.sleep(1) sem.release() print('Worker %i released semaphore' % n)gevent.joinall([gevent.spawn(worker, i) for i in range(0, 4)])

data=local()def f1():    data.x = 1    print('协程1:{}'.format(data.x))def f2():    data.x = 2    try:        print('协程2:{}'.format(data.x))    except AttributeError:print('xisnotvisible')gevent.joinall([    gevent.spawn(f1),    gevent.spawn(f2)])