# 不加鎖:并發(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é)可能有疑問:既然加鎖會讓運(yùn)行變成串行,那么我在start之后立即使用join,就不用加鎖了啊,也是串行的效果啊
# 沒錯:在start之后立刻使用jion,肯定會將100個任務(wù)的執(zhí)行變成串行,毫無疑問,最終n的結(jié)果也肯定是0,是安全的,但問題是
# start后立即join:任務(wù)內(nèi)的所有代碼都是串行執(zhí)行的,而加鎖,只是加鎖的部分即修改共享數(shù)據(jù)的部分是串行的
# 單從保證數(shù)據(jù)安全方面,二者都可以實現(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 #耗時是多么的恐怖
'''

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)值為True，所有阻塞池的線程激活進(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('鏈接超時')
        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('驗證成功',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)入一個元組,元組的第一個元素是優(yōu)先級(通常是數(shù)字,也可以是非數(shù)字之間的比較),數(shù)字越小優(yōu)先級越高
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)先級越高,優(yōu)先級高的優(yōu)先出隊):
(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)造一個優(yōu)先級隊列,其中maxsize是一個整數(shù),用于設(shè)置可以放入隊列的項目數(shù)量的上限.一旦達(dá)到這個上限，插入就會阻塞，直到隊列中有項目被消耗。如果maxsize小于或等于0，則隊列長度為無窮大。

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)過排序后取到的索引為0的那個元素,一般條目是(優(yōu)先級數(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()在一個空的隊列對象中被調(diào)用時,會拋出異常

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()在一個滿的隊列對象中被調(diào)用時,會拋出異常

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

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

Queue.full() # return True if full 

當(dāng)隊列為滿返回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).

將一個項放入隊列。如果可選參數(shù)block為true并且timeout為None(默認(rèn)值)，則在必要時阻塞，直到有空閑槽可用。如果參數(shù)timeout是一個正數(shù)，它最多阻塞timeout秒，如果在這段時間內(nèi)沒有可用的空閑槽，則會引發(fā)Full異常。否則(block為false)，如果有空閑槽可用，則將一個項目放入隊列中，否則引發(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).

從隊列中移除并返回一個項。如果可選參數(shù)block為true并且timeout為None(默認(rèn)值)，則在必要時阻塞，直到有可用的項。如果timeout為正數(shù)，則最多阻塞timeout秒，如果在該時間內(nèi)沒有可用項，則拋出Empty異常。否則(block為false)，如果一個項目可用，則返回那個項目，否則引發(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.

提供了兩種方法來支持追蹤進(jìn)入隊列的任務(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)入隊列的任務(wù)已完成。并且被隊列生產(chǎn)者使用。對于每個用于獲取任務(wù)的get()，后續(xù)對task_done()的調(diào)用都會告訴隊列任務(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)前正被阻塞，它將在所有項都被處理完時恢復(fù)(這意味著對于每個已經(jīng)put()到隊列中的項都接收到task_done()調(diào)用)。

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

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

Queue.join() 

阻塞,直到queue被消費完畢

# 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，立即返回，并不會等待池內(nèi)的任務(wù)執(zhí)行完畢
但不管wait參數(shù)為何值，整個程序都會等到所有任務(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類是Executor的子類，它使用一個進(jìn)程池來異步執(zhí)行調(diào)用。ProcessPoolExecutor會調(diào)用多進(jìn)程模塊，這允許它避開全局解釋器鎖，但也意味著只能執(zhí)行和返回可pickle的對象。

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ù)量不超過參數(shù)max_workers的子進(jìn)程,如果max_workers為None或未給出，則默認(rèn)值為機(jī)器上的處理器數(shù)。如果max_workers小于或等于0，則會拋出異常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拿到的是一個future對象obj，需要用obj.result()拿到結(jié)果