# 不加鎖:并發(fā)執(zhí)行,速度快,數(shù)據(jù)不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print('%s is running' %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1

if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
'''

# 不加鎖:未加鎖部分并發(fā)執(zhí)行,加鎖部分串行執(zhí)行,速度慢,數(shù)據(jù)安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    # 未加鎖的代碼并發(fā)運(yùn)行
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    # 加鎖的代碼串行運(yùn)行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
'''

# 有的同學(xué)可能有疑問(wèn):既然加鎖會(huì)讓運(yùn)行變成串行,那么我在start之后立即使用join,就不用加鎖了啊,也是串行的效果啊
# 沒(méi)錯(cuò):在start之后立刻使用jion,肯定會(huì)將100個(gè)任務(wù)的執(zhí)行變成串行,毫無(wú)疑問(wèn),最終n的結(jié)果也肯定是0,是安全的,但問(wèn)題是
# start后立即join:任務(wù)內(nèi)的所有代碼都是串行執(zhí)行的,而加鎖,只是加鎖的部分即修改共享數(shù)據(jù)的部分是串行的
# 單從保證數(shù)據(jù)安全方面,二者都可以實(shí)現(xiàn),但很明顯是加鎖的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1

if __name__ == '__main__':
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗時(shí)是多么的恐怖
'''

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print('\033[41m%s 拿到A鎖\033[0m' %self.name)

        mutexB.acquire()
        print('\033[42m%s 拿到B鎖\033[0m' %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('\033[43m%s 拿到B鎖\033[0m' %self.name)
        time.sleep(2)

        mutexA.acquire()
        print('\033[44m%s 拿到A鎖\033[0m' %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()

'''
Thread-1 拿到A鎖
Thread-1 拿到B鎖
Thread-1 拿到B鎖
Thread-2 拿到A鎖
然后就卡住，死鎖了
'''

from threading import Thread,Semaphore
import threading
import time
# def func():
#     if sm.acquire():
#         print (threading.currentThread().getName() + ' get semaphore')
#         time.sleep(2)
#         sm.release()
def func():
    sm.acquire()
    print('%s get sm' %threading.current_thread().getName())
    time.sleep(3)
    sm.release()
if __name__ == '__main__':
    sm=Semaphore(5)
    for i in range(23):
        t=Thread(target=func)
        t.start()

event.isSet()：返回event的狀態(tài)值；

event.wait()：如果 event.isSet()==False將阻塞線程；

event.set()： 設(shè)置event的狀態(tài)值為T(mén)rue，所有阻塞池的線程激活進(jìn)入就緒狀態(tài)， 等待操作系統(tǒng)調(diào)度；

event.clear()：恢復(fù)event的狀態(tài)值為False。

from threading import Thread,Event
import threading
import time,random
def conn_mysql():
    count=1
    while not event.is_set():
        if count > 3:
            raise TimeoutError('鏈接超時(shí)')
        print('<%s>第%s次嘗試鏈接' % (threading.current_thread().getName(), count))
        event.wait(0.5)
        count+=1
    print('<%s>鏈接成功' %threading.current_thread().getName())

def check_mysql():
    print('\033[45m[%s]正在檢查mysql\033[0m' % threading.current_thread().getName())
    time.sleep(random.randint(2,4))
    event.set()
if __name__ == '__main__':
    event=Event()
    conn1=Thread(target=conn_mysql)
    conn2=Thread(target=conn_mysql)
    check=Thread(target=check_mysql)

    conn1.start()
    conn2.start()
    check.start()

import threading

def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()
def condition_func():

    ret = False
    inp = input('>>>')
    if inp == '1':
        ret = True

    return ret

def run(n):
    con.acquire()
    con.wait_for(condition_func)
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

from threading import Timer

def hello():
    print("hello, world")

t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

from threading import Timer
import random,time

class Code:
    def __init__(self):
        self.make_cache()

    def make_cache(self,interval=5):
        self.cache=self.make_code()
        print(self.cache)
        self.t=Timer(interval,self.make_cache)
        self.t.start()

    def make_code(self,n=4):
        res=''
        for i in range(n):
            s1=str(random.randint(0,9))
            s2=chr(random.randint(65,90))
            res+=random.choice([s1,s2])
        return res

    def check(self):
        while True:
            inp=input('>>: ').strip()
            if inp.upper() ==  self.cache:
                print('驗(yàn)證成功',end='\n')
                self.t.cancel()
                break

if __name__ == '__main__':
    obj=Code()
    obj.check()

import queue

q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
結(jié)果(先進(jìn)先出):
first
second
third
'''

import queue

q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
結(jié)果(后進(jìn)先出):
third
second
first
'''

import queue

q=queue.PriorityQueue()
#put進(jìn)入一個(gè)元組,元組的第一個(gè)元素是優(yōu)先級(jí)(通常是數(shù)字,也可以是非數(shù)字之間的比較),數(shù)字越小優(yōu)先級(jí)越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))

print(q.get())
print(q.get())
print(q.get())
'''
結(jié)果(數(shù)字越小優(yōu)先級(jí)越高,優(yōu)先級(jí)高的優(yōu)先出隊(duì)):
(10, 'b')
(20, 'a')
(30, 'c')
'''

Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

構(gòu)造一個(gè)優(yōu)先級(jí)隊(duì)列,其中maxsize是一個(gè)整數(shù),用于設(shè)置可以放入隊(duì)列的項(xiàng)目數(shù)量的上限.一旦達(dá)到這個(gè)上限，插入就會(huì)阻塞，直到隊(duì)列中有項(xiàng)目被消耗。如果maxsize小于或等于0，則隊(duì)列長(zhǎng)度為無(wú)窮大。

The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).

首先檢索最低值的條目(最低值的條目是指列表經(jīng)過(guò)排序后取到的索引為0的那個(gè)元素,一般條目是(優(yōu)先級(jí)數(shù)字,數(shù)據(jù))這種元組的形式

exception queue.Empty
Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.

當(dāng)表示非阻塞的get()或get_nowait()在一個(gè)空的隊(duì)列對(duì)象中被調(diào)用時(shí),會(huì)拋出異常

exception queue.Full
Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.

當(dāng)表示非阻塞的put()或put_nowait()在一個(gè)滿的隊(duì)列對(duì)象中被調(diào)用時(shí),會(huì)拋出異常

Queue.qsize()
Queue.empty() #return True if empty  

當(dāng)隊(duì)列為空返回True

Queue.full() # return True if full 

當(dāng)隊(duì)列為滿返回True

Queue.put(item, block=True, timeout=None)
Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).

將一個(gè)項(xiàng)放入隊(duì)列。如果可選參數(shù)block為true并且timeout為None(默認(rèn)值)，則在必要時(shí)阻塞，直到有空閑槽可用。如果參數(shù)timeout是一個(gè)正數(shù)，它最多阻塞timeout秒，如果在這段時(shí)間內(nèi)沒(méi)有可用的空閑槽，則會(huì)引發(fā)Full異常。否則(block為false)，如果有空閑槽可用，則將一個(gè)項(xiàng)目放入隊(duì)列中，否則引發(fā)Full異常(在這種情況下，timeout被忽略)。

Queue.put_nowait(item)
Equivalent to put(item, False).

Queue.get(block=True, timeout=None)
Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).

從隊(duì)列中移除并返回一個(gè)項(xiàng)。如果可選參數(shù)block為true并且timeout為None(默認(rèn)值)，則在必要時(shí)阻塞，直到有可用的項(xiàng)。如果timeout為正數(shù)，則最多阻塞timeout秒，如果在該時(shí)間內(nèi)沒(méi)有可用項(xiàng)，則拋出Empty異常。否則(block為false)，如果一個(gè)項(xiàng)目可用，則返回那個(gè)項(xiàng)目，否則引發(fā)Empty異常(在這種情況下，timeout被忽略)。

Queue.get_nowait()
Equivalent to get(False).

Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.

提供了兩種方法來(lái)支持追蹤進(jìn)入隊(duì)列的任務(wù)是否已被生產(chǎn)者的守護(hù)線程完全處理。

Queue.task_done()
Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

假定先前進(jìn)入隊(duì)列的任務(wù)已完成。并且被隊(duì)列生產(chǎn)者使用。對(duì)于每個(gè)用于獲取任務(wù)的get()，后續(xù)對(duì)task_done()的調(diào)用都會(huì)告訴隊(duì)列任務(wù)的處理已經(jīng)完成。

If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).

如果join()當(dāng)前正被阻塞，它將在所有項(xiàng)都被處理完時(shí)恢復(fù)(這意味著對(duì)于每個(gè)已經(jīng)put()到隊(duì)列中的項(xiàng)都接收到task_done()調(diào)用)。

Raises a ValueError if called more times than there were items placed in the queue.

如果調(diào)用次數(shù)超過(guò)放入隊(duì)列的項(xiàng)數(shù)，將引發(fā)ValueError。

Queue.join() 

阻塞,直到queue被消費(fèi)完畢

# 1 介紹
concurrent.futures模塊提供了高度封裝的異步調(diào)用接口
ThreadPoolExecutor：線程池，提供異步調(diào)用
ProcessPoolExecutor: 進(jìn)程池，提供異步調(diào)用
Both implement the same interface, which is defined by the abstract Executor class.

# 2 基本方法
# submit(fn, *args, **kwargs)
異步提交任務(wù)

# map(func, *iterables, timeout=None, chunksize=1) 
取代for循環(huán)submit的操作

# shutdown(wait=True) 
相當(dāng)于進(jìn)程池的pool.close()+pool.join()操作
wait=True，等待池內(nèi)所有任務(wù)執(zhí)行完畢回收完資源后才繼續(xù)
wait=False，立即返回，并不會(huì)等待池內(nèi)的任務(wù)執(zhí)行完畢
但不管wait參數(shù)為何值，整個(gè)程序都會(huì)等到所有任務(wù)執(zhí)行完畢
submit和map必須在shutdown之前

# result(timeout=None)
取得結(jié)果

# add_done_callback(fn)
回調(diào)函數(shù)

# 介紹
The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.

ProcessPoolExecutor類(lèi)是Executor的子類(lèi)，它使用一個(gè)進(jìn)程池來(lái)異步執(zhí)行調(diào)用。ProcessPoolExecutor會(huì)調(diào)用多進(jìn)程模塊，這允許它避開(kāi)全局解釋器鎖，但也意味著只能執(zhí)行和返回可pickle的對(duì)象。

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)
An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.
能夠異步調(diào)用數(shù)量不超過(guò)參數(shù)max_workers的子進(jìn)程,如果max_workers為None或未給出，則默認(rèn)值為機(jī)器上的處理器數(shù)。如果max_workers小于或等于0，則會(huì)拋出異常ValueError

# 用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ProcessPoolExecutor(max_workers=3)

    futures=[]
    for i in range(11):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)
    print('+++>')
    for future in futures:
        print(future.result())

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit
回調(diào)函數(shù)
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def get_page(url):
    print('<進(jìn)程%s> get %s' %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}

def parse_page(res):
    res=res.result()
    print('<進(jìn)程%s> parse %s' %(os.getpid(),res['url']))
    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
    with open('db.txt','a') as f:
        f.write(parse_res)


if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]

    # p=Pool(3)
    # for url in urls:
    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)
    # p.close()
    # p.join()

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一個(gè)future對(duì)象obj，需要用obj.result()拿到結(jié)果