Python 三程三器的那些事

裝飾器

一、什麼是裝飾器

• 裝飾器本質是函數，用來給其餘函數添加新的功能
• 特色：不修改調用方式、不修改源代碼

二、裝飾器的做用

• 裝飾器做用：本質是函數（裝飾其餘函數）就是爲其餘函數添加其餘功能
• 裝飾器必須準尋得原則：
  • 不能修改被裝飾函數的源代碼、不能修改被裝飾函數的調用方式
• 實現裝飾器知識儲備：
  • 函數即「變量」
  • 高階函數
  • 嵌套函數 高階函數+潛逃函數=》裝飾器

三、使用高階函數模仿裝飾器功能

　　　　1.定義：把一個函數名當作實參傳給另外一個函數
　　　　2.返回值中包含函數名
　　　　3.下面使用高階函數雖然能夠實現裝飾器的一些功能，可是違反了裝飾器不能改變調用方式的原則，
　　　　     之前使用bar()如今將調用方式改編成了test1(bar)就是將bar的函數名當作變量傳給了test1()

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import time

def timer(func):
    start_time = time.time()
    func()
    print '函數執行時間爲', time.time() - start_time

def test():
    print '開始執行test'
    time.sleep(3)
    print 'test執行結束'


timer(test)
'''
開始執行test
test執行結束
函數執行時間爲 3.00332999229
'''

改變了調用方式

　　　　4.高階函數——不修改高階函數的調用方式增長新的功能（可是沒法傳參數）
　　　　注：bar = test2(bar) 等價於：@timer從新將函數名bar賦值，將原函數bar的內存地址當作實參傳遞該函數test2()，再將test2()賦值給bar

import time
def bar():
    time.sleep(3)
    print("in the bar")
def test2(func):
    print(func)
    return func
bar = test2(bar)
bar()

不改變調用方式

　　　　5.嵌套函數
　　　　　　嵌套函數：在一個函數中嵌套另外一個函數，並在函數內部調用

def foo():
    print("in the foo")
    def bar():
        print("in the bar")
    bar()
foo()

嵌套函數

四、可以適應90%的業務需求

• 在裝飾器中 @timer等價於 test1=timer(test1)
• 在timer()函數中返回值是return deco
• 因此timer(test1)做用是將函數test1內存地址當作參數傳遞給timer()
• timer() 函數最後將運行後的函數deco內存地址做爲返回值返回
• test1=timer(test1)做用就是將將deco函數內存地址賦值給test1,因此最後運行test1()就至關於運行deco()
• 因此最後調用時給test2()傳入參數就至關於給deco傳入參數

import time
def timer(func):   #timer(test1)  func=test1
    def deco(*args,**kwargs):
        start_time = time.time()
        func(*args,**kwargs)      #run test1
        stop_time = time.time()
        print("running time is %s"%(stop_time-start_time))
    return deco
@timer     # test1=timer(test1)
def test1():
    time.sleep(3)
    print("in the test1")
@timer
def test2(name):
    print("in the test2",name)
test1()
test2("tom")

裝飾器1

五、對特定網頁進行身份驗證

import time
user,passwd = 'aaa','123'
def auth(func):
    def wrapper(*args,**kwargs):
        username = input("Username:").strip()
        password = input("Password:").strip()
        if user == username and password == passwd:
            print("User has passed authentication")
            res = func(*args,**kwargs)   #這裏執行func()至關於執行調用的函數如home()
            return res          #爲了得到home()函數返回值，能夠將執行結果賦值給res而後返回print(home())結果是"from home"而不是"None"了
        else:
            exit("Invalid username or password")
    return wrapper
def index():
    print("welcome to index page")
@auth
def home():
    print("welcome to home page")
    return "from home"
@auth
def bbs():
    print("welcome to bbs page")
index()
print(home())   #在這裏調用home()至關於調用wrapper()
bbs()

裝飾器2

六、實現對不一樣網頁不一樣方式的身份認證

• @auth(auth_type="local")代碼做用
• 在上面的代碼中使用@auth至關於先將home函數的內存地址當作變量傳入auth()函數，執行結果後home()至關於wrapper()
• 而在這裏驗證的時候猶豫@auth(auth_type="local")中有()括號，那麼就至關於將執行auth()函數並且是將auth_type="local當作參數傳入到auth()函數執行
• 因此outer_wrapper函數也會執行，outer_wrapper函數的執行結果返回的就是wrapper()函數的內存地址
• 因此最終結果一樣是執行home()函數就至關於執行wrapper函數
• 可是有所不一樣的是着這個版本中咱們能夠在外層的auth函數中傳入新的參數幫組咱們根據需求判斷 

import time
user,passwd = 'aaa','123'
def auth(auth_type):
    print("auth func:",auth_type)
    def outer_wrapper(func):
        def wrapper(*args, **kwargs):
            print("wrapper func args:", *args, **kwargs)
            if auth_type == "local":
                username = input("Username:").strip()
                password = input("Password:").strip()
                if user == username and passwd == password:
                    print("\033[32;1mUser has passed authentication\033[0m")
                    res = func(*args, **kwargs)  # from home
                    print("---after authenticaion ")
                    return res
                else:
                    exit("\033[31;1mInvalid username or password\033[0m")
            elif auth_type == "ldap":
                print("搞毛線ldap,不會。。。。")

        return wrapper
    return outer_wrapper

def index():
    print("welcome to index page")
@auth(auth_type="local") # home = wrapper()
def home():
    print("welcome to home  page")
    return "from home"

@auth(auth_type="ldap")
def bbs():
    print("welcome to bbs  page")

index()
print(home()) #wrapper()
bbs()

裝飾器3

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import time
def auth(auth_type):
    print("auth func:",auth_type)
    def outer_wrapper(func):
        def wrapper(*args, **kwargs):
            print("wrapper func args:", *args, **kwargs)
            print('運行前')
            func(*args, **kwargs)
            print('運行後')
        return wrapper
    return outer_wrapper

@auth(auth_type="local") # home = wrapper()
def home():
    print("welcome to home  page")
    return "from home"
home()

三級裝飾器簡寫

七、使用閉包實現裝飾器功能

　　閉包概念：

• 在一個外函數中定義了一個內函數，內函數裏運用了外函數的臨時變量，而且外函數的返回值是內函數的引用，這樣就構成了一個閉包
• 通常狀況下，在咱們認知當中，若是一個函數結束，函數的內部全部東西都會釋放掉，還給內存，局部變量都會消失。
• 可是閉包是一種特殊狀況，若是外函數在結束的時候發現有本身的臨時變量未來會在內部函數中用到，就把這個臨時變量綁定給了內部函數，而後本身再結束。

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import time

def timer(func):                    #timer(test1)  func=test1
    def deco(*args,**kwargs):       # # 函數嵌套
        start_time = time.time()
        func(*args,**kwargs)        # 跨域訪問，引用了外部變量func   (func實質是函數內存地址)
        stop_time = time.time()
        print "running time is %s"%(stop_time-start_time)
    return deco                    # 內層函數做爲外層函數返回值

def test(name):
    print "in the test2",name
    time.sleep(2)

test = timer(test)   # 等價於 ==》 @timer語法糖
test("tom")
'''
運行結果：
in the test2 tom
running time is 2.00302696228
'''

閉包實現裝飾器功能

 

生成器

一、什麼是生成器

• 生成器就是一個特殊的迭代器
• 一個有yield關鍵字的函數就是一個生成器
• 生成器是這樣一個函數，它記住上一次返回時在函數體中的位置。
• 對生成器函數的第二次（或第 n 次）調用跳轉至該函數中間，而上次調用的全部局部變量都保持不變。

二、定義

• 生成器，即生成一個容器。
• 在Python中，一邊循環，一邊計算的機制，稱爲生成器。
• 生成器能夠理解爲一種數據類型，這種數據類型自動實現了迭代器協議（其餘數據類型須要調用本身的內置iter（）方法或__iter__()的內置函數），
• 因此，生成器就是一個可迭代對象。

三、生成器哪些場景應用

• 生成器是一個概念，咱們日常寫代碼可能用的並很少，可是python源碼大量使用
• 好比咱們tornado框架就是基於 生成器+協程
• 在咱們代碼中使用舉例
• 好比咱們要生成一百萬個數據，若是用生成器很是節省空間，用列表浪費大量空間

import time
t1 = time.time()
g = (i for i in range(100000000))
t2 = time.time()
lst = [i for i in range(100000000)]
t3 = time.time()
print('生成器時間：',t2 - t1)  # 生成器時間： 0.0
print('列表時間：',t3 - t2)    # 列表時間： 5.821957349777222

四、生成器的做用

• 經過列表生成式，咱們能夠直接建立一個列表，可是，受到內存限制，列表容量確定是有限的。
• 並且，建立一個包含100萬個元素的列表，不只佔用很大的存儲空間，若是咱們僅僅須要訪問前面幾個元素，那後面絕大多數元素佔用的空間都白白浪費了。
• 因此，若是列表元素能夠按照某種算法推算出來，那咱們是否能夠在循環的過程當中不斷推算出後續的元素呢？
• 這樣就沒必要建立完整的list，從而節省大量的空間。在Python中，這種一邊循環一邊計算的機制，稱爲生成器：generator。
• 要建立一個generator，有不少種方法，第一種方法很簡單，只要把一個列表生成式的[]改爲()，就建立了一個generator：

print( [i*2 for i in range(10)] )             #列表生成式： [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
print( (i*2 for i in range(10)) )             #生  成  器： <generator object <genexpr> at 0x005A3690>

• 咱們能夠直接打印出list的每個元素，但咱們怎麼打印出generator的每個元素呢？
• 若是要一個一個打印出來，能夠經過next()函數得到generator的下一個返回值：

g = (i*2 for i in range(10))
print( g.__next__() )               # 0
print( g.__next__() )               # 2

 五、生成器工做原理

• 生成器是這樣一個函數，它記住上一次返回時在函數體中的位置。
• 對生成器函數的第二次（或第 n 次）調用跳轉至該函數中間，而上次調用的全部局部變量都保持不變。
• 生成器不只「記住」了它數據狀態；生成器還「記住」了它在流控制構造（在命令式編程中，這種構造不僅是數據值）中的位置。
• 生成器是一個函數，並且函數的參數都會保留。
• 迭代到下一次的調用時，所使用的參數都是第一次所保留下的，便是說，在整個全部函數調用的參數都是第一次所調用時保留的，而不是新建立的

六、yield生成器運行機制

• 在Python中，yield就是這樣的一個生成器。
• 當你問生成器要一個數時，生成器會執行，直至出現 yield 語句，生成器把yield 的參數給你，以後生成器就不會往下繼續運行。
• 當你問他要下一個數時，他會從上次的狀態開始運行，直至出現yield語句，把參數給你，以後停下。如此反覆
• 在python中，當你定義一個函數，使用了yield關鍵字時，這個函數就是一個生成器
• 它的執行會和其餘普通的函數有不少不一樣，函數返回的是一個對象，而不是你日常所用return語句那樣，能獲得結果值。若是想取得值，那得調用next()函數
• 每當調用一次迭代器的next函數，生成器函數運行到yield之處，返回yield後面的值且在這個地方暫停，全部的狀態都會被保持住，直到下次next函數被調用，或者碰到異常循環退出。

def fib(max_num):
    a,b = 1,1
    while a < max_num:
        yield b
        a,b=b,a+b

g = fib(10)               #生成一個生成器：[1，2, 3, 5, 8, 13]
print(g.__next__())       #第一次調用返回：1
print(list(g))            #把剩下元素變成列表：[2, 3, 5, 8, 13]

yield實現fib數

七、yield實現單線程下的併發效果

• yield至關於 return 返回一個值，而且記住這個返回的位置，下次迭代時，代碼從yield的下一條語句開始執行。
• send() 和next()同樣，都能讓生成器繼續往下走一步（下次遇到yield停），但send()能傳一個值，這個值做爲yield表達式總體的結果

def consumer(name):
    print("%s 準備吃包子啦!" %name)
    while True:
       baozi = yield
       print("包子[%s]來了,被[%s]吃了!" %(baozi,name))
c = consumer("Tom")
c.__next__()
b1 = "韭菜餡包子"
c.send(b1)

# c.send(b1)做用：
# c.send()的做用是給yied的傳遞一個值，而且每次調用c.send()的同時自動調用一次__next__
'''運行結果：
Tom 準備吃包子啦!
包子[韭菜餡包子]來了,被[Tom]吃了!
'''

一次調用

import time
def consumer(name):
    print("%s 準備吃包子啦!" %name)
    while True:
       baozi = yield
       print("包子[%s]來了,被[%s]吃了!" %(baozi,name))
def producer(name):
    c = consumer('A')
    c2 = consumer('B')
    c.__next__()
    c2.__next__()
    print("老子開始準備作包子啦!")
    for i in range(10):
        time.sleep(1)
        print("作了2個包子!")
        c.send(i)
        c2.send(i)
producer("alex")

'''運行結果：
A 準備吃包子啦!
B 準備吃包子啦!
老子開始準備作包子啦!
作了2個包子!
包子[0]來了,被[A]吃了!
包子[0]來了,被[B]吃了!
作了2個包子!
包子[1]來了,被[A]吃了!
包子[1]來了,被[B]吃了!
作了2個包子!
包子[2]來了,被[A]吃了!
包子[2]來了,被[B]吃了!
作了2個包子!
包子[3]來了,被[A]吃了!
包子[3]來了,被[B]吃了!
作了2個包子!
包子[4]來了,被[A]吃了!
包子[4]來了,被[B]吃了!
作了2個包子!
包子[5]來了,被[A]吃了!
包子[5]來了,被[B]吃了!
'''

for 循環調用

 

迭代器

一、什麼是迭代器

• 迭代器是訪問集合內元素的一種方法
• 老是從集合內第一個元素訪問，直到全部元素都被訪問過結束，當調用 __next__而元素返回會引起一個，StopIteration異常
• 有兩個方法：_iter_ _next_
• _iter_ : 返回迭代器自身
• _next_: 返回下一個元素

二、定義：

• 迭代器是訪問集合內元素的一種方式。迭代器對象從集合的第一個元素開始訪問，直到全部的元素都被訪問一遍後結束。

三、迭代器和可迭代對象

• 凡是可做用於for循環的對象都是可迭代的（Iterable）類型；
• 凡是可做用於next()函數的對象都是迭代器（Iterator）類型，它們表示一個惰性計算的序列；
• 集合數據類型如list、dict、str等是可迭代的但不是迭代器，不過能夠經過iter()函數得到一個Iterator對象。
• Python的for循環本質上就是經過不斷調用next()函數實現的
• 總結： 一個實現了__iter__方法的對象是可迭代的，一個實現next方法的對象是迭代器

四、迭代器的兩個方法

• 迭代器僅是一容器對象，它實現了迭代器協議。它有兩個基本方法
  • __next__方法：返回容器的下一個元素
  • __iter__方法：返回迭代器自身
• 迭代器是訪問集合內元素的一種方式。迭代器對象從集合的第一個元素開始訪問，直到全部的元素都被訪問一遍後結束。
• __iter__方法會返回一個迭代器(iterator)，所謂的迭代器就是具備next方法的對象。
• 在調用next方法時，迭代器會返回它的下一個值，若是next方法被調用，但迭代器中沒有值能夠返就會引起一個StopIteration異常

a = iter([1,2,])              #生成一個迭代器
print(a.__next__())
print(a.__next__())
print(a.__next__())           #在這一步會引起  「StopIteration」 的異常

五、判斷是迭代器和可迭代對象

　　注：列表，元組，字典是可迭代的但不是迭代器

from collections import Iterable
print(isinstance([],Iterable))                               #True
print(isinstance({},Iterable))                               #True
print(isinstance((),Iterable))                               #True
print(isinstance("aaa",Iterable))                           #True
print(isinstance((x for x in range(10)),Iterable))          #True

相關代碼

六、列表不是迭代器，只有生成器是迭代器

from collections import Iterator
t = [1,2,3,4]
print(isinstance(t,Iterator))           #False
t1 = iter(t)
print(isinstance(t1,Iterator))          #True

相關代碼

七、自定義迭代器

#! /usr/bin/env python
# -*- coding: utf-8 -*-
class MyRange(object):
    def __init__(self, n):
        self.idx = 0
        self.n = n

    def __iter__(self):
        return self

    def next(self):
        if self.idx < self.n:
            val = self.idx
            self.idx += 1
            return self.n[val]
        else:
            raise StopIteration()


l = [4,5,6,7,8]
obj = MyRange(l)
print obj.next()      # 4
print obj.next()      # 5
print obj.next()      # 6

自定義迭代器

八、迭代器與生成器

#! /usr/bin/env python
# -*- coding: utf-8 -*
l = [1,2,3,4,5]                  # 列表是一個可迭代對象，不是一個迭代器
print dir(l)                     # 因此 l 中有 __iter__() 方法，沒有 __next__()方法
iter_obj = l.__iter__()          # __iter__()方法返回迭代器對象自己(這個迭代器對象就會有 next 方法了)

print '###################################\n'
print iter_obj.next()     # 1
print iter_obj.next()     # 2
print iter_obj.next()     # 3

相關代碼

 

進程與線程的簡介

一、什麼是進程(process)？（進程是資源集合）

二、進程是資源分配的最小單位（ 內存、cpu、網絡、io）

三、一個運行起來的程序就是一個進程

• 什麼是程序（程序是咱們存儲在硬盤裏的代碼）
• 硬盤（256G）、內存條（8G）
• 當咱們雙擊圖標，打開程序的時候，實際上就是經過I/O操做（讀寫）內存條裏面
• 內存條就是咱們所指的資源

CPU分時

• CPU比你的手速快多了，分時處理每一個線程，可是因爲太快然你以爲每一個線程都是獨佔cpu
• cpu是計算，只有時間片到了，獲取cpu，線程真正執行
• 當你想使用 網絡、磁盤等資源的時候，須要cpu的調度
• 進程具備獨立的內存空間，因此沒有辦法相互通訊

進程如何通訊

• 進程queue(父子進程通訊)
• pipe（同一程序下兩個進程通訊）
• managers（同一程序下多個進程通訊）
• RabbitMQ、redis等（不一樣程序間通訊）

爲何須要進程池

• 一次性開啓指定數量的進程
• 若是有十個進程，有一百個任務，一次能夠處理多少個（一次性只能處理十個）
• 防止進程開啓數量過多致使服務器壓力過大

二、定義：進程是資源分配最小單位

• 當一個可執行程序被系統執行（分配內存資源）就變成了一個進程
• 程序並不能單獨運行，只有將程序裝載到內存中，系統爲它分配資源才能運行，這種執行的程序就稱之爲進程
• 程序和進程的區別就在於：程序是指令的集合，它是進程運行的靜態描述文本；進程是程序的一次執行活動，屬於動態概念
• 在多道編程中，咱們容許多個程序同時加載到內存中，在操做系統的調度下，能夠實現併發地執行。
• 進程的出現讓每一個用戶感受到本身獨享CPU，所以，進程就是爲了在CPU上實現多道編程而提出的。
• 進程之間有本身獨立的內存，各進程之間不能相互訪問
• 建立一個新線程很簡單，建立新進程須要對父進程進行復制
• 多道編程： 在計算機內存中同時存放幾道相互獨立的程序，他們共享系統資源，相互穿插運行
• 單道編程： 計算機內存中只容許一個的程序運行

三、進程併發性：

• 在一個系統中，同時會存在多個進程被加載到內存中，同處於開始到結束之間的狀態
• 對於一個單CPU系統來講，程序同時處於運行狀態只是一種宏觀上的概念
• 他們雖然都已經開始運行，但就微觀而言，任意時刻，CPU上運行的程序只有一個
• 因爲操做系統分時，讓每一個進程都以爲本身獨佔CPU等資源
• 注：若是是多核CPU（處理器）其實是能夠實現正在乎義的同一時間點有多個線程同時運行

四、線程併發性：

• 操做系統將時間劃分爲不少時間段，儘量的均勻分配給每個線程。
• 獲取到時間片的線程被CPU執行，其餘則一直在等待，因此微觀上是走走停停，宏觀上都在運行。
  • 多核CPU狀況：　　　　　　　　　　
    • 若是你的程序的線程數少於CPU的核心數，且系統此時沒有其餘進程同時運行，那麼這個程序的每一個線程會享有一個CPU，
    • 當同時運行的線程數多於CPU核心數時，CPU會採用必定的調度算法每隔一段時間就將這些線程調入或調出CPU
    • 以確保每一個線程都能分享一部分CPU時間，實現多線程併發。

五、有了進程爲何還要線程？

1.進程優勢：

• 提供了多道編程，讓咱們感受咱們每一個人都擁有本身的CPU和其餘資源，能夠提升計算機的利用率

2. 進程的兩個重要缺點

• 進程只能在一個時間幹一件事，若是想同時幹兩件事或多件事，進程就無能爲力了。
• 進程在執行的過程當中若是阻塞，即便進程中有些工做不依賴於輸入的數據，也將沒法執行（例如等待輸入，整個進程就會掛起）。
• 例如，咱們在使用qq聊天， qq作爲一個獨立進程若是同一時間只能幹一件事，那他如何實如今同一時刻 即能監聽鍵盤輸入、又能監聽其它人給你發的消息
• 你會說，操做系統不是有分時麼？分時是指在不一樣進程間的分時呀
• 即操做系統處理一會你的qq任務，又切換到word文檔任務上了，每一個cpu時間片分給你的qq程序時，你的qq仍是隻能同時幹一件事呀

六、什麼是線程(thread)（線程是操做系統最小的調度單位）

• 定義：
  • 線程是操做系統調度的最小單位
  • 它被包含在進程之中，是進程中的實際運做單位
  • 進程自己是沒法本身執行的，要操做cpu，必須建立一個線程，線程是一系列指令的集合
• 線程是操做系統可以進行運算調度的最小單位。它被包含在進程之中，是進程中的實際運做單位
• 一條線程指的是進程中一個單一順序的控制流，一個進程中能夠併發多個線程，每條線程並行執行不一樣的任務
•  不管你啓多少個線程，你有多少個cpu, Python在執行的時候會淡定的在同一時刻只容許一個線程運行
• 進程自己是沒法本身執行的，要操做cpu，必須建立一個線程，線程是一系列指令的集合
• 全部在同一個進程裏的線程是共享同一塊內存空間的，不一樣進程間內存空間不一樣
• 同一個進程中的各線程能夠相互訪問資源，線程能夠操做同進程中的其餘線程，但進程僅能操做子進程
• 兩個進程想通訊，必需要經過一箇中間代理
• 對主線程的修改可能回影響其餘子線程，對主進程修改不會影響其餘進程由於進程間內存相互獨立，可是同一進程下的線程共享內存

七、進程和線程的區別

• 啓動一個線程比啓動一個進程快，運行速度沒有可比性。
• 先有一個進程而後纔能有線程。
  • 進程包含線程
  • 線程共享內存空間
  • 進程內存是獨立的（不可互相訪問）
  • 進程能夠生成子進程，子進程之間互相不能互相訪問（至關於在父級進程克隆兩個子進程）
  • 在一個進程裏面線程之間能夠交流。兩個進程想通訊，必須經過一箇中間代理來實現
  • 建立新線程很簡單，建立新進程須要對其父進程進行克隆。
  • 一個線程能夠控制或操做同一個進程裏面的其它線程。但進程只能操做子進程。
  • 父進程能夠修改不影響子進程，但不能修改。
  • 線程能夠幫助應用程序同時作幾件事

八、進程和程序的區別

• 程序只是一個普通文件，是一個機器代碼指令和數據的集合，因此，程序是一個靜態的實體
• 而進程是程序運行在數據集上的動態過程，進程是一個動態實體，它應建立而產生，應調度執行因等待資 源或事件而被處於等待狀態，因完成任務而被撤消
• 進程是系統進行資源分配和調度的一個獨立單位
• 一個程序對應多個進程，一個進程爲多個程序服務（二者之間是多對多的關係）
• 一個程序執行在不一樣的數據集上就成爲不一樣的進程，能夠用進程控制塊來惟一地標識每一個進程

 

多線程

Python多線程編程中經常使用方法：

• join()方法：若是一個線程或者在函數執行的過程當中調用另外一個線程，而且但願待其完成操做後才能執行，那麼在調用線程的時就可使用被調線程的join方法join([timeout]) timeout：可選參數，線程運行的最長時間
• isAlive()方法：查看線程是否還在運行
• getName()方法：得到線程名
• setDaemon()方法：主線程退出時，須要子線程隨主線程退出，則設置子線程的setDaemon()

GIL全局解釋器鎖：

• 在python全局解釋器下，保證同一時間只有一個線程運行
• 防止多個線程都修改數據

線程鎖（互斥鎖）：

• GIL鎖只能保證同一時間只能有一個線程對某個資源操做，但當上一個線程還未執行完畢時可能就會釋放GIL，其餘線程就能夠操做了
• 線程鎖本質把線程中的數據加了一把互斥鎖
• mysql中共享鎖 & 互斥鎖
  • mysql共享鎖：共享鎖，全部線程都能讀，而不能寫
  • mysql排它鎖：排它，任何線程讀取這個這個數據的權利都沒有
• 加上線程鎖以後全部其餘線程，讀都不能讀這個數據
• 有了GIL全局解釋器鎖爲何還須要線程鎖
  • 由於cpu是分時使用的

一、線程2種調用方式：直接調用， 繼承式調用

import threading
import time

def sayhi(num):                                   # 定義每一個線程要運行的函數
    print("running on number:%s" % num)
    time.sleep(3)

#一、target=sayhi ：sayhi是定義的一個函數的名字
#二、args=(1,)    ： 括號內寫的是函數的參數
t1 = threading.Thread(target=sayhi, args=(1,))    # 生成一個線程實例
t2 = threading.Thread(target=sayhi, args=(2,))    # 生成另外一個線程實例

t1.start()                                        # 啓動線程
t2.start()                                        # 啓動另外一個線程

print(t1.getName())                               # 獲取線程名
print(t2.getName())

直接調用

import threading
import time

class MyThread(threading.Thread):
    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):#定義每一個線程要運行的函數
        print("running on number:%s" %self.num)
        time.sleep(3)

if __name__ == '__main__':
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

繼承式調用

二、for循環同時啓動多個線程

• 說明：下面利用for循環同時啓動50個線程並行執行，執行時間是3秒而不是全部線程執行時間的總和

import threading
import time

def sayhi(num): #定義每一個線程要運行的函數
    print("running on number:%s" %num)
    time.sleep(3)
for i in range(50):
    t = threading.Thread(target=sayhi,args=('t-%s'%i,))
    t.start()

for循環啓動多個線程

三、t.join()： 實現全部線程都執行結束後再執行主線程

• 說明：在4中雖然能夠實現50個線程同時併發執行，可是主線程不會等待子線程結束在這裏咱們可使用t.join()指定等待某個線程結束的結果

import threading
import time
start_time = time.time()

def sayhi(num): #定義每一個線程要運行的函數
    print("running on number:%s" %num)
    time.sleep(3)

t_objs = []    #將進程實例對象存儲在這個列表中
for i in range(50):
    t = threading.Thread(target=sayhi,args=('t-%s'%i,))
    t.start()          #啓動一個線程，程序不會阻塞
    t_objs.append(t)
print(threading.active_count())    #打印當前活躍進程數量
for t in t_objs: #利用for循環等待上面50個進程所有結束
    t.join()     #阻塞某個程序
print(threading.current_thread())    #打印執行這個命令進程

print("----------------all threads has finished.....")
print(threading.active_count())
print('cost time:',time.time() - start_time)

t.join() 主線程等待子線程

四、setDaemon(): 守護線程，主線程退出時，須要子線程隨主線程退出

import threading
import time
start_time = time.time()

def sayhi(num): #定義每一個線程要運行的函數
    print("running on number:%s" %num)
    time.sleep(3)
for i in range(50):
    t = threading.Thread(target=sayhi,args=('t-%s'%i,))
    t.setDaemon(True)  #把當前線程變成守護線程，必須在t.start()前設置
    t.start()          #啓動一個線程，程序不會阻塞
print('cost time:',time.time() - start_time)

守護線程

• 注：由於剛剛建立的線程是守護線程，因此主線程結束後子線程就結束了，運行時間不是3秒而是0.01秒

五、GIL鎖和用戶鎖（Global Interpreter Lock 全局解釋器鎖）

• 全局解釋器鎖：保證同一時間僅有一個線程對資源有操做權限
  • 做用：在一個進程內，同一時刻只能有一個線程經過GIL鎖 被CUP調用，切換條件：I/O操做、固定時間(系統決定)
  • 說明：python多線程中GIL鎖只是在CPU操做時（如：計算）纔是串行的，其餘都是並行的，因此比串行快不少
  • 爲了解決不一樣線程同時訪問同一資源時，數據保護問題，而產生了GIL
  • GIL在解釋器的層面限制了程序在同一時間只有一個線程被CPU實際執行，而無論你的程序裏實際開了多少條線程
  • 爲了解決這個問題，CPython本身定義了一個全局解釋器鎖，同一時間僅僅有一個線程能夠拿到這個數據
  • python之因此會產生這種很差的情況是由於python啓用一個線程是調用操做系統原生線程，就是C接口
  • 可是這僅僅是CPython這個版本的問題，在PyPy，中就沒有這種缺陷
• 用戶鎖：線程鎖(互斥鎖Mutex)  ：當前線程還未操做完成前其餘全部線程都沒法對其操做，即便已經釋放了GIL鎖
• 在有GIL鎖時爲什麼還須要用戶鎖
  • GIL鎖只能保證同一時間只能有一個線程對某個資源操做，但當上一個線程還未執行完畢時可能就會釋放GIL，其餘線程就能夠操做了
• 線程鎖的原理
  • 當一個線程對某個資源進行CPU計算的操做時加一個線程鎖，只有當前線程計算完成主動釋放鎖，其餘線程才能對其操做
  • 這樣就能夠防止還未計算完成，釋放GIL鎖後其餘線程對這個資源操做致使混亂問題

import time
import threading
lock = threading.Lock()          #1 生成全局鎖
def addNum():
    global num                  #2 在每一個線程中都獲取這個全局變量
    print('--get num:',num )
    time.sleep(1)
    lock.acquire()              #3 修改數據前加鎖
    num  -= 1                   #4 對此公共變量進行-1操做
    lock.release()              #5 修改後釋放

用戶鎖使用舉例

在有GIL的狀況下執行 count = count + 1 會出錯緣由解析，用線程鎖解決方法

# 1）第一步：count = 0   count初始值爲0
        # 2）第二步：線程1要執行對count加1的操做首先申請GIL全局解釋器鎖
        # 3）第三步：調用操做系統原生線程在操做系統中執行
        # 4）第四步：count加1還未執行完畢，時間到了被要求釋放GIL
        # 5）第五步：線程1釋放了GIL後線程2此時也要對count進行操做，此時線程1還未執行完，因此count仍是0
        # 6）第六步：線程2此時拿到count = 0後也要對count進行加1操做，假如線程2執行很快，一次就完成了
        #    count加1的操做，那麼count此時就從0變成了1
        # 7）第七步：線程2執行完加1後就賦值count=1並釋放GIL
        # 8）第八步：線程2執行完後cpu又交給了線程1，線程1根據上下文繼續執行count加1操做，先拿到GIL
        #    鎖，完成加1操做，因爲線程1先拿到的數據count=0，執行完加1後結果仍是1
        # 9）第九步：線程1將count=1在次賦值給count並釋放GIL鎖，此時連個線程都對數據加1，可是值最終是1

報錯緣由分析

• 使用線程鎖解決上面問題的原理
  • 在GIL鎖中再加一個線程鎖，線程鎖是用戶層面的鎖
  • 線程鎖就是一個線程在對數據操做前加一把鎖，防止其餘線程複製或者操做這個數據
  • 只有這個線程對數據操做完畢後纔會釋放這個鎖，其餘線程才能操做這個數據
• 定義一個線程鎖很是簡單隻用三步

1 >> lock = threading.Lock()                        #定義一把鎖
2 >> lock.acquire()                                   #對數據操做前加鎖防止數據被另外一線程操做
3 >> lock.release()                                     #對數據操做完成後釋放鎖

六、死鎖

• 死鎖定義：
  • 兩個以上的進程或線程在執行過程當中，因爭奪資源而形成的一種互相等待的現象若無外力做用，它們都將沒法推動去。
• 死鎖舉例：
  • 啓動5個線程，執行run方法,假如thread1首先搶到了A鎖，此時thread1沒有釋放A鎖，緊接着執行代碼mutexB.acquire()，搶到了B鎖，
  • 在搶B鎖時候，沒有其餘線程與thread1爭搶，由於A鎖沒有釋放，其餘線程只能等待
  • thread1執行完func1函數，而後執行func2函數，此時thread1拿到B鎖，而後執行time.sleep(2)，此時不會釋放B鎖
  • 在thread1執行func2的同時thread2開始執行func1獲取到了A鎖，而後繼續要獲取B鎖
  • 不幸的是B鎖還被thread1佔用，thread1佔用B鎖時還須要同時獲取A鎖才能向下執行，可是此時發現A鎖已經被thread2暫用，這樣就死鎖了

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(2):
        t=MyThread()
        t.start()

# 運行結果：輸出下面結果後程序卡死，再也不向下進行了
# Thread-1 拿到A鎖
# Thread-1 拿到B鎖
# Thread-1 拿到B鎖
# Thread-2 拿到A鎖

產生死鎖代碼

七、遞歸鎖：lock = threading.RLock()  解決死鎖問題

• 遞歸鎖的做用是同一線程中屢次請求同一資源，可是不會參數死鎖。
• 這個RLock內部維護着一個Lock和一個counter變量，counter記錄了acquire的次數，從而使得資源能夠被屢次require。
• 直到一個線程全部的acquire都被release，其餘的線程才能得到資源。

from threading import Thread,Lock,RLock
import time

mutexA=mutexB=RLock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到A鎖' %self.name)

        mutexB.acquire()
        print('%s 拿到B鎖' %self.name)
        mutexB.release()

        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('%s 拿到B鎖' % self.name)
        time.sleep(0.1)
        mutexA.acquire()
        print('%s 拿到A鎖' % self.name)
        mutexA.release()

        mutexB.release()

if __name__ == '__main__':
    for i in range(5):
        t=MyThread()
        t.start()
# 下面是運行結果：不會產生死鎖
# Thread-1 拿到A鎖
# Thread-1 拿到B鎖
# Thread-1 拿到B鎖
# Thread-1 拿到A鎖
# Thread-2 拿到A鎖
# Thread-2 拿到B鎖
# Thread-2 拿到B鎖
# Thread-2 拿到A鎖
# Thread-4 拿到A鎖
# Thread-4 拿到B鎖
# Thread-4 拿到B鎖
# Thread-4 拿到A鎖
# Thread-3 拿到A鎖
# Thread-3 拿到B鎖
# Thread-3 拿到B鎖
# Thread-3 拿到A鎖
# Thread-5 拿到A鎖
# Thread-5 拿到B鎖
# Thread-5 拿到B鎖
# Thread-5 拿到A鎖

若是使用RLock代替Lock，則不會發生死鎖

八、Semaphore(信號量)

• 互斥鎖 同時只容許一個線程更改數據，而Semaphore是同時容許必定數量的線程更改數據
• 好比廁全部3個坑，那最多隻容許3我的上廁所，後面的人只能等裏面有人出來了才能再進去
• 做用就是同一時刻容許運行的線程數量

# import threading,time
# def run(n):
#     semaphore.acquire()
#     time.sleep(1)
#     print("run the thread: %s\n" %n)
#     semaphore.release()
# 
# if __name__ == '__main__':
#     semaphore  = threading.BoundedSemaphore(5) #最多容許5個線程同時運行
#     for i in range(22):
#         t = threading.Thread(target=run,args=(i,))
#         t.start()
# 
# while threading.active_count() != 1:
#     pass #print threading.active_count()
# else:
#     print('----all threads done---')


# 代碼結果說明：這裏能夠清晰看到運行時0-4是同時運行的沒有順序，並且是前五個，
# 表示再semaphore這個信號量的定義下程序同時僅能執行5個線程

信號量舉例

九、events總共就只有四個方法

1. event.set()          : # 設置標誌位
2. event.clear()       : # 清除標誌位
3. event.wait()        : # 等待標誌被設定
4. event.is_set()     : # 判斷標誌位是否被設定

import time,threading

event = threading.Event()
#第一：寫一個紅綠燈的死循環
def lighter():
    count = 0
    event.set()               #1先設置爲綠燈
    while True:
        if count > 5 and count <10:      #2改爲紅燈
            event.clear()          #3把標誌位清了
            print("red light is on.....")
        elif count > 10:
            event.set()            #4再設置標誌位，變綠燈
            count = 0
        else:
            print("green light is on.....")
        time.sleep(1)
        count += 1

#第二：寫一個車的死循環
def car(name):
    while True:
        if event.is_set():         #設置了標誌位表明綠燈
            print("[%s] is running"%name)
            time.sleep(1)
        else:
            print('[%s] sees red light, waiting......'%name)
            event.wait()
            print('[%s] green light is on,start going.....'%name)

light = threading.Thread(target=lighter,)
light.start()
car1 = threading.Thread(target=car,args=("Tesla",))
car1.start()

events（紅綠燈例子）

 

進程 

• 多線程和多進程各自應用場景
  • I/O操做不佔用CPU（從硬盤，網路讀入數據等）
  • 計算佔用CPU，這種狀況最好不用多線程
  • python多線程不適合CPU密集型的任務，適合I/O密集型的任務
  • python的多進程適合CPU密集型任務
• 一次性起多個進程，並在進程中調用線程

import multiprocessing,time,threading

#3 被多線程調用的函數
def thread_run():
    print(threading.get_ident())   #打印線程id號
    time.sleep(2)

#2 被多進程調用的函數，以及在這個函數中起一個進程
def run(name):
    time.sleep(2)
    print("hello",name)
    t = threading.Thread(target=thread_run,)  #在進程調用的函數中啓用一個線程
    t.start()

#1 一次性啓動多個進程
if __name__ == '__main__':
    for i in range(10):
        p = multiprocessing.Process(target=run,args=('bob %s'%i,)) #啓用一個多線程
        p.start()

一次性起多個進程，並在進程中調用線程

• 進程間互相訪問數據的三種方法
  • 注：不一樣進程間內存是不共享的，因此互相之間不能訪問對方數據
• 在父進程中定義隊列q，使用父進程啓用一個子進程，子進程中沒法操做父進程的q

from multiprocessing import Process
import queue
import threading
def f():
    q.put([42, None, 'hello'])
 
if __name__ == '__main__':
    q = queue.Queue()              #1 在父進程中定義一個隊列實例q
    # p = threading.Thread(target=f,)    #在線程程中就能夠相互訪問，線程中內存共享
    p = Process(target=f,)        #2 在父進程中起一個子進程 p，在子進程中使用父進程的q會報錯
    p.start()
    print(q.get())
    p.join()

子進程沒法訪問父進程數據舉例

• 利用Queues實現父進程到子進程（或子進程間）的數據傳遞

• • 咱們之前學的queue是線程queue.Queue()只有在同一個進程的線程間才能訪問
  • 若是兩個進程間想要通訊必需要使用進程Queue，用法和多線程的相同
  • queue.Queue()是線程q不能夠傳遞給子進程，可是Queue是進程q，父進程會將進程q克隆了一份給子進程
  • 既然是兩個q爲何在子進程中在q中放入一個數據在父進程中能夠取出來呢？ 其實緣由是這樣的：
    • 子進程向q中放入數據的時候，用pickle序列化將數據放到一箇中間地方（翻譯），翻譯又把子進程放
    • 入的數據用pickle反序列化給父進程，父進程就能夠訪問這個q了，這樣就實現了進程間的數據通訊了
    • 在多線程中兩個線程能夠修改同一份數據，而Queue僅僅實現了進程間的數據傳遞

from multiprocessing import Process, Queue

def f(qq):  # 將符進程中的q傳遞過來叫qq
    qq.put([42, None, 'hello'])  # 此時子進程就可使用符進程中的q

if __name__ == '__main__':
    q = Queue()  # 使用Queue()在父進程中定義一個隊列實例q
    p = Process(target=f, args=(q,))  # 在父進程中起一個子進程 p，將父進程剛定義的q傳遞給子進程p
    p.start()
    print(q.get())
    p.join()

# 運行結果： [42, None, 'hello']

Queues實現父子進程間傳遞數據

• 使用管道pipe實現兩個進程間數據傳遞
  • 說明：其實pip實現進程間通訊就好像一條電話線同樣，一個在電話線這頭髮送，一個在電話線那頭接收

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])  # 3 子進程發送數據，就像socket同樣
    print("son process recv:", conn.recv())
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    # 1 生成一個管道實例，實例一輩子成就會生成兩個返回對象，一個是管道這頭，一個是管道那頭
    p = Process(target=f, args=(child_conn,))  # 2 啓動一個子進程將管道其中一頭傳遞給子進程
    p.start()
    print(parent_conn.recv())  # 4 父進程收消息 # prints "[42, None, 'hello']"
    parent_conn.send('i am parent process')
    p.join()

# 運行結果：
# [42, None, 'hello']
# son process recv: i am parent process

pip實現進程間通訊

• Managers實現不少進程間數據共享

• • 說明：manager實質和Queue同樣，啓用是個線程其實就是將字典或者列表copy十份

from multiprocessing import Process, Manager
import os

def f(d, l):
    d[1] = '1'  # 是個進程對字典放入的是同一個值，因此看上去效果不明顯
    l.append(os.getpid())  # 將這是個進程的進程id放入列表中


if __name__ == '__main__':
    with Manager() as manager:  # 1 將Manager()賦值給manager
        d = manager.dict()  # 2 定義一個能夠在多個進程間能夠共享的字典
        l = manager.list(range(5))  # 3 定義一個能夠在多個進程間能夠共享的列表，默認寫五個數據
        p_list = []
        for i in range(10):  # 生成是個進程
            p = Process(target=f, args=(d, l))  # 將剛剛生成的可共享字典和列表傳遞給子進程
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
        print(d)
        print(l)

managers實現進程間數據共享

• 進程之間須要鎖的緣由
  • 說明：雖然每一個進程是獨立運行的，可是他們共享同一塊屏幕，若是你們都在屏幕打數據就會打亂了

from multiprocessing import Process, Lock
def f(l, i):
    l.acquire()                     #一個進程要打印數據時先鎖定
    print('hello world', i)
    l.release()                     #打印完畢後就釋放這把鎖
if __name__ == '__main__':
    lock = Lock()                   #先生成一把鎖
    for num in range(5):
        Process(target=f, args=(lock, num)).start()

# 運行結果：
# hello world 4
# hello world 0
# hello world 2
# hello world 3
# hello world 1

進程鎖

• 進程池
• 進程池的做用就是限制同一時間能夠啓動進程的=數量
• 進程池內部維護一個進程序列，當使用時，則去進程池中獲取一個進程，若是進程池序列中沒有可供使用的進那麼程序就會等待，直到進程池中有可用進程爲止。
• 進程池中有兩個方法：
  • apply： 多個進程異步執行，一個一個的執行
  • apply_async： 多個進程同步執行，同時執行多個進程

from  multiprocessing import Process,Pool
import time,os
def foo(i):
    time.sleep(2)
    print("in the process",os.getpid()) #打印子進程的pid
    return i+100

def call(arg):
    print('-->exec done:',arg,os.getpid())

if __name__ == '__main__':
    pool = Pool(3)                      #進程池最多容許5個進程放入進程池
    print("主進程pid：",os.getpid())     #打印父進程的pid
    for i in range(10):
        #用法1 callback做用是指定只有當Foo運行結束後就執行callback調用的函數,父進程調用的callback函數
        pool.apply_async(func=foo, args=(i,),callback=call)

        #用法2 串行 啓動進程不在用Process而是直接用pool.apply()
        # pool.apply(func=foo, args=(i,))

    print('end')
    pool.close()    #關閉pool
    pool.join()     #進程池中進程執行完畢後再關閉，若是註釋，那麼程序直接關閉。

進程池

殭屍進程

• 殭屍進程定義
  • 殭屍進程產生的緣由就是父進程產生子進程後，子進程先於父進程退出
  • 可是父進程因爲種種緣由，並無處理子進程發送的退出信號，那麼這個子進程就會成爲殭屍進程。
• 用python寫一個殭屍進程

#!/usr/bin/env python
#coding=utf8
 
import os, sys, time
#產生子進程
pid = os.fork()
 
if pid == 0:
    #子進程退出
    sys.exit(0)
#父進程休息30秒
time.sleep(30)
# 先產生一個子進程，子進程退出，父進程休息30秒,那就會產生一個殭屍進程

defunct.py

[root@linux-node4 ~]# ps -ef| grep defunct
root     110401  96083  0 19:11 pts/2    00:00:00 python defunct.py
root     110402 110401  0 19:11 pts/2    00:00:00 [python] <defunct>
root     110406  96105  0 19:11 pts/3    00:00:00 grep --color=auto defunct

ps -ef| grep defunct 在linux下查看殭屍進程

 

協程(Coroutine) 

一、什麼是協程（進入上一次調用的狀態）

• 協程，又稱微線程，纖程，協程是一種用戶態的輕量級線程。
• 線程的切換會保存到CPU的棧裏，協程擁有本身的寄存器上下文和棧，
• 協程調度切換時，將寄存器上下文和棧保存到其餘地方，在切回來的時候，恢復先前保存的寄存器上下文和棧
• 協程能保留上一次調用時的狀態（即全部局部狀態的一個特定組合），每次過程重入時，就至關於進入上一次調用的狀態
• 協程最主要的做用是在單線程的條件下實現併發的效果，但實際上仍是串行的（像yield同樣）

二、協程的好處

• 無需線程上下文切換的開銷（能夠理解爲協程切換就是在不一樣函數間切換，不用像線程那樣切換上下文CPU）
• 不須要多線程的鎖機制，由於只有一個線程，也不存在同時寫變量衝突
• 用法：最簡單的方法是多進程+協程，既充分利用多核，又充分發揮協程的高效率，可得到極高的性能。

三、協程缺點

• 沒法利用多核資源：協程的本質是個單線程,它不能同時將 單個CPU 的多個核用上,協程須要和進程配合才能運行在多CPU上
• 線程阻塞（Blocking）操做（如IO時）會阻塞掉整個程序

四、使用yield實現協程相同效果

import time
import queue

def consumer(name):
    print("--->starting eating baozi...")
    while True:
        new_baozi = yield  # 只要遇到yield程序就返回，yield還能夠接收數據
        print("[%s] is eating baozi %s" % (name, new_baozi))
        time.sleep(1)

def producer():
    r = con.__next__()  # 直接調用消費者的__next__方法
    r = con2.__next__()  # 函數裏面有yield第一次加括號調用會變成一個生成器函數不執行，運行next才執行
    n = 0
    while n < 5:
        n += 1
        con.send(n)  # send恢復生成器同時並傳遞一個值給yield
        con2.send(n)
        print("\033[32;1m[producer]\033[0m is making baozi %s" % n)

if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    p = producer()

yield模擬實現協程效果

五、協程爲什麼能處理大併發1：Greenlet遇到I/O手動切換

• 協程之因此快是由於遇到I/O操做就切換（最後只有CPU運算）
• 這裏先演示用greenlet實現手動的對各個協程之間切換
• 其實Gevent模塊僅僅是對greenlet的再封裝，將I/O間的手動切換變成自動切換

from greenlet import greenlet

def test1():
    print(12)       #4 gr1會調用test1()先打印12
    gr2.switch()    #5 而後gr2.switch()就會切換到gr2這個協程
    print(34)       #8 因爲在test2()切換到了gr1，因此gr1又從上次中止的位置開始執行
    gr2.switch()    #9 在這裏又切換到gr2，會再次切換到test2()中執行

def test2():
    print(56)       #6 啓動gr2後會調用test2()打印56
    gr1.switch()    #7 而後又切換到gr1
    print(78)       #10 切換到gr2後會接着上次執行，打印78

gr1 = greenlet(test1)    #1 啓動一個協程gr1
gr2 = greenlet(test2)    #2 啓動第二個協程gr2
gr1.switch()             #3 首先gr1.switch() 就會去執行gr1這個協程

Greenlet遇到I/O手動切換

六、協程爲什麼能處理大併發2：Gevent遇到I/O自動切換

• Gevent 是一個第三方庫，能夠輕鬆經過gevent實現併發同步或異步編程
• 在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程
• Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度。
• Gevent原理是隻要遇到I/O操做就會自動切換到下一個協程

七、Gevent實現簡單的自動切換小例子

• 注：在Gevent模仿I/O切換的時候，只要遇到I/O就會切換，哪怕gevent.sleep(0)也要切換一次

import gevent

def func1():
    print('\033[31;1m第一次打印\033[0m')
    gevent.sleep(2)          # 爲何用gevent.sleep()而不是time.sleep()由於是爲了模仿I/O
    print('\033[31;1m第六次打印\033[0m')

def func2():
    print('\033[32;1m第二次打印\033[0m')
    gevent.sleep(1)
    print('\033[32;1m第四次打印\033[0m')

def func3():
    print('\033[32;1m第三次打印\033[0m')
    gevent.sleep(1)
    print('\033[32;1m第五次打印\033[0m')

gevent.joinall([            # 將要啓動的多個協程放到event.joinall的列表中，便可實現自動切換
    gevent.spawn(func1),    # gevent.spawn(func1)啓動這個協程
    gevent.spawn(func2),
    gevent.spawn(func3),
])

# 運行結果：
# 第一次打印
# 第二次打印
# 第三次打印
# 第四次打印
# 第五次打印
# 第六次打印

Gevent實現簡單的自動切換小例子

八、使用Gevent實現併發下載網頁與串行下載網頁時間比較

from urllib import request
import gevent,time
from gevent import monkey
monkey.patch_all()      #把當前程序全部的I/O操做給我單獨作上標記

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

#1 併發執行部分
time_binxing = time.time()
gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])
print("並行時間：",time.time()-time_binxing)

#2 串行部分
time_chuanxing = time.time()
urls = [
        'https://www.python.org/',
        'https://www.yahoo.com/',
        'https://github.com/',
                                        ]
for url in urls:
    f(url)
print("串行時間：",time.time()-time_chuanxing)

# 注：爲何要在文件開通使用monkey.patch_all()
# 1. 由於有不少模塊在使用I / O操做時Gevent是沒法捕獲的，因此爲了使Gevent可以識別出程序中的I / O操做。
# 2. 就必須使用Gevent模塊的monkey模塊，把當前程序全部的I / O操做給我單獨作上標記
# 3.使用monkey作標記僅用兩步便可：
      第一步(導入monkey模塊)：  from gevent import monkey
      第二步(聲明作標記)    ：   monkey.patch_all()

並行串行時間比較

• 說明：monkey.patch_all()猴子補丁做用

• • 用過gevent就會知道,會在最開頭的地方gevent.monkey.patch_all();
  • 做用是把標準庫中的thread/socket等給替換掉.這樣咱們在後面使用socket的時候能夠跟日常同樣使用,無需修改任何代碼,可是它變成非阻塞的了.

九、經過gevent本身實現單線程下的多socket併發

import gevent
from gevent import socket,monkey     #下面使用的socket是Gevent的socket，實際測試monkey沒用
# monkey.patch_all()

def server(port):
    s = socket.socket()
    s.bind(('0.0.0.0',port))
    s.listen(5)
    while True:
        cli,addr = s.accept()
        gevent.spawn(handle_request,cli)

def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            print('recv:',data)
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__=='__main__':
    server(8001)

server端

import socket
HOST = 'localhost'    # The remote host
PORT = 8001           # The same port as used by the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
    msg = bytes(input(">>:"),encoding="utf8").strip()
    if len(msg) == 0:continue
    s.sendall(msg)
    data = s.recv(1024)
    print('Received', repr(data))
s.close()

client端

十、協程本質原理

• 協程1經過os去讀一個file，這個時候就是一個io操做，在調用os的接口前，就會有一個列表
• 協程1的這個操做就會被註冊到這個列表中，而後就切換到其餘協程去處理；
• 等待os拿到要讀file後，也會把這個文件句柄放在這個列表中
• 而後等待在切換到協程1的時候，協程1就能夠直接從列表中拿到數據，這樣就能夠實現不阻塞了
• epoll返回給協程的任務列表在內核態，協程在用戶態，用戶態協程是不能直接訪問內核態的任務列表的，因此須要拷貝整個內核態的任務列表到用戶態，供協程去訪問和查詢

十一、epoll處理 I/O 請求原理

• epoll() 中內核則維護一個鏈表，epoll_wait 直接檢查鏈表是否是空就知道是否有文件描述符準備好了。
• 在內核實現中 epoll 是根據每一個 sockfd 上面的與設備驅動程序創建起來的回調函數實現的。
• 某個 sockfd 上的事件發生時，與它對應的回調函數就會被調用，來把這個 sockfd 加入鏈表，其餘處於「空閒的」狀態的則不會。
• epoll上面鏈表中獲取文件描述，這裏使用內存映射（mmap）技術， 避免了複製大量文件描述符帶來的開銷
• 內存映射（mmap）：內存映射文件，是由一個文件到一塊內存的映射，將沒必要再對文件執行I/O操做

十二、select處理協程

• 拷貝全部的文件描述符給協程，不論這些任務的是否就緒，都會被返回
• 那麼協程就只能for循環去查找本身的文件描述符，也就是任務列表，select的兼容性很是好，支持linux和windows

1三、select、epool、pool

• I/O的實質是什麼？
  • I/O的實質是將硬盤中的數據，或收到的數據實現從內核態 copy到 用戶態的過程
  • 本文討論的背景是Linux環境下的network IO。
  • 好比微信讀取本地硬盤的過程
    • 微信進程會發送一個讀取硬盤的請求----》操做系統
    • 只有內核纔可以讀取硬盤中的數據---》數據返回給微信程序（看上去就好像是微信直接讀取）
• 用戶態 & 內核態
  • 系統空間分爲兩個部分，一部分是內核態，一部分是用戶態的部分
  • 內核態：內核態的空間資源只有操做系統可以訪問
  • 用戶態：咱們寫的普通程序使用的空間

  

• select
  • 只能處理1024個鏈接（每個請求均可以理解爲一個鏈接）
  • 不能告訴用戶程序，哪個鏈接是活躍的
• pool
  • 只是取消了最大1024個活躍的限制
  • 不能告訴用戶程序，哪個鏈接是活躍的
• epool
  • 不只取消了1024這個最大鏈接限制
  • 並且能告訴用戶程序哪個是活躍的

Python進程池和線程池(ThreadPoolExecutor&ProcessPoolExecutor) 

• 簡介 參考官網
  • Python標準庫爲咱們提供了threading和multiprocessing模塊編寫相應的多線程/多進程代碼
  • 可是當項目達到必定的規模，頻繁建立/銷燬進程或者線程是很是消耗資源的，這個時候咱們就要編寫本身的線程池/進程池，以空間換時間。
  • 但從Python3.2開始，標準庫爲咱們提供了concurrent.futures模塊，它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個類，
  • 實現了對threading和multiprocessing的進一步抽象，對編寫線程池/進程池提供了直接的支持。

• Executor和Future

　　　　1. Executor

• • concurrent.futures模塊的基礎是Exectuor，Executor是一個抽象類，它不能被直接使用。
  • 可是它提供的兩個子類ThreadPoolExecutor和ProcessPoolExecutor倒是很是有用
  • 咱們能夠將相應的tasks直接放入線程池/進程池，不須要維護Queue來操心死鎖的問題，線程池/進程池會自動幫咱們調度。

　　　　2. Future

• • Future你能夠把它理解爲一個在將來完成的操做，這是異步編程的基礎，
  • 傳統編程模式下好比咱們操做queue.get的時候，在等待返回結果以前會產生阻塞，cpu不能讓出來作其餘事情，
  • 而Future的引入幫助咱們在等待的這段時間能夠完成其餘的操做。
• ThreadPoolExecutor(線程池)

from concurrent.futures import ThreadPoolExecutor
import time
def return_future_result(message):
    time.sleep(2)
    return message
pool = ThreadPoolExecutor(max_workers=2)                 # 建立一個最大可容納2個task的線程池
future1 = pool.submit(return_future_result, ("hello"))  # 往線程池裏面加入一個task
future2 = pool.submit(return_future_result, ("world"))  # 往線程池裏面加入一個task

print(future1.done())      # 判斷task1是否結束
time.sleep(3)
print(future2.done())      # 判斷task2是否結束
print(future1.result())    # 查看task1返回的結果
print(future2.result())    # 查看task2返回的結果

# 運行結果：
# False   # 這個False與下面的True會等待3秒
# True    # 後面三個輸出都是一塊兒打出來的
# hello
# world

使用submit來操做線程池/進程池

import concurrent.futures
import urllib.request
URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # Start the load operations and mark each future with its URL
    # future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}   # 這一句至關於下面for循環獲取的字典
    future_to_url = {}
    for url in URLS:
        future_to_url[executor.submit(load_url,url,60)] = url      # {'future對象':'url'}   future對象做爲key，url做爲value
    for future in concurrent.futures.as_completed(future_to_url): # as_completed返回已經有返回結果的future對象
        url = future_to_url[future]                                # 經過future對象獲取對應的url
        try:
            data = future.result()                                 # 獲取future對象的返回結果
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

使用for循環使用線程池，並將future對象加入字典中

from concurrent.futures import ThreadPoolExecutor

# 建立線程池
executor = ThreadPoolExecutor(10)

def test_function(num1,num2):
    return "%s + %s = %s"%(num1,num2,num1+num2)
result_iterators = executor.map(test_function,[1,2,3],[5,6,7])

for result in result_iterators:
    print(result)

# 1 + 5 = 6
# 2 + 6 = 8
# 3 + 7 = 10

最簡單map舉例 

import concurrent.futures
import urllib.request

URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']

def load_url(url):
    with urllib.request.urlopen(url, timeout=60) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future_dic = {}
    for url, data in zip(URLS, executor.map(load_url, URLS)):
        print('%r page is %d bytes' % (url, len(data)))
        future_dic[url] = data     # {'url':'執行結果'}   url做爲key，執行結果做爲value

# 'http://httpbin.org' page is 13011 bytes
# 'http://example.com/' page is 1270 bytes
# 'https://api.github.com/' page is 2039 bytes

使用map同時獲取多個頁面中的數據

使用線程池、進程池、協程向多個url併發獲取頁面數據比較 

• 特色：

• • 進程：啓用進程很是浪費資源
  • 線程：線程多，而且在阻塞過程當中沒法執行其餘任務
  • 協程：gevent只用起一個線程，當請求發出去後gevent就無論,永遠就只有一個線程工做，誰先回來先處理
• 使用for循環串行拿取頁面數據（第四：性能最差）

import requests
url_list = [
    'https://www.baidu.com',
    'http://dig.chouti.com/',
]

for url in url_list:
    result = requests.get(url)
    print(result.text)

使用for循環串行拿取頁面數據（效果最差）

• 進程池實現併發（第三）
  • 缺點：啓用進程很是浪費資源

import requests
from concurrent.futures import ProcessPoolExecutor

def fetch_request(url):
    result = requests.get(url)
    print(result.text)

url_list = [
    'https://www.baidu.com',
    'https://www.google.com/',         #google頁面會卡住，知道頁面超時後這個進程才結束
    'http://dig.chouti.com/',          #chouti頁面內容會直接返回，不會等待Google頁面的返回
]

if __name__ == '__main__':
    pool = ProcessPoolExecutor(10)        # 建立線程池
    for url in url_list:
        pool.submit(fetch_request,url)    # 去線程池中獲取一個進程，進程去執行fetch_request方法
    pool.shutdown(False)

使用進程池實現併發

• 線程池實現併發（第二）
  • 缺點： 建立一個新線程將消耗大量的計算資源，而且在阻塞過程當中沒法執行其餘任務。
  • 例： 好比線程池中10個線程同時去10個url獲取數據，當數據還沒來時這些線程所有都在等待，不作事。

import requests
from concurrent.futures import ThreadPoolExecutor

def fetch_request(url):
    result = requests.get(url)
    print(result.text)

url_list = [
    'https://www.baidu.com',
    'https://www.google.com/',         #google頁面會卡住，知道頁面超時後這個進程才結束
    'http://dig.chouti.com/',          #chouti頁面內容會直接返回，不會等待Google頁面的返回
]

pool = ThreadPoolExecutor(10)            # 建立一個線程池，最多開10個線程
for url in url_list:
    pool.submit(fetch_request,url)       # 去線程池中獲取一個線程，線程去執行fetch_request方法

pool.shutdown(True)                      # 主線程本身關閉，讓子線程本身拿任務執行

使用線程池實現併發

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch_async(url):
    response = requests.get(url)
    return response.text

def callback(future):
    print(future.result())

url_list = ['http://www.github.com', 'http://www.bing.com']
pool = ThreadPoolExecutor(5)
for url in url_list:
    v = pool.submit(fetch_async, url)
    v.add_done_callback(callback)
pool.shutdown(wait=True)

多線程+回調函數執行

• 協程：微線程實現異步（第一：性能最好）
  • 特色 ：gevent只用起一個線程，當請求發出去後gevent就無論,永遠就只有一個線程工做，誰先回來先處理

import gevent
import requests
from gevent import monkey

monkey.patch_all()

# 這些請求誰先回來就先處理誰
def fetch_async(method, url, req_kwargs):
    response = requests.request(method=method, url=url, **req_kwargs)
    print(response.url, response.content)

# ##### 發送請求 #####
gevent.joinall([
    gevent.spawn(fetch_async, method='get', url='https://www.python.org/', req_kwargs={}),
    gevent.spawn(fetch_async, method='get', url='https://www.google.com/', req_kwargs={}),
    gevent.spawn(fetch_async, method='get', url='https://github.com/', req_kwargs={}),
])

協程：微線程實現異步