Python高级编程、asyncio和高并发爬虫

Alan

6 年前

一、Python中一切皆对象

函数和类也是对象，属于python的一等公民

1. 赋值给一个变量

2. 可以添加到集合对象中

3. 可以作为参数传递给函数

4. 可以当做函数的返回值

type、object和class的关系

object 是最顶层基类，type是一个类，同时也是一个对象

>>> type.__bases__
(<class 'object'>,)
>>> type(object)
<class 'type'>
>>> type(type)
<class 'type'>

>>> type.__bases__

(<class 'object'>,)

>>> type(object)

>>> type(type)

Python中的内置类型

对象的三个特征：身份(id())、类型(type())、值

类型：None(全局只有一个)、数值（int, float, complex, bool）、迭代类型、序列类型（list, bytes, bytearray, memoryview, range, tuple, str, array）、映射(dict)、集合（set, frozenset）、上下文管理类型（with）、其它类型

二、魔法函数

__init__等以双下划线开头、双下划线结尾的系统内置函数

非数学运算

字符串表示
__repr__、__str__
集合、序列相关
__len__、__getitem__、__setitem__、__delitem__、__contains__
迭代相关
__iter__、__next__
可调用
__call__
with上下文管理器
__enter__、__exit__
数值转换
__abs__、__bool__、__int__、__float__、__hash__、__index__
元类相关
__new__、__init__
属性相关
__getattr__、 __setattr__、__getattribute__、setattribute__、__dir__
属性描述符
__get__、__set__、 __delete__
协程
__await__、__aiter__、__anext__、__aenter__、__aexit__

数学运算

一元运算符
__neg__（-）、__pos__（+）、__abs__
二元运算符
__lt__(<)、 __le__ <= 、 __eq__ == 、 __ne__ != 、 __gt__ > 、 __ge__ >=
算术运算符
__add__ + 、 __sub__ – 、 __mul__ * 、 __truediv__ / 、 __floordiv__ // 、 __
mod__ % 、 __divmod__ divmod() 、 __pow__ ** 或 pow() 、 __round__ round()
反向算术运算符
__radd__ 、 __rsub__ 、 __rmul__ 、 __rtruediv__ 、 __rfloordiv__ 、 __rmod__ 、
__rdivmod__ 、 __rpow__
增量赋值算术运算符
__iadd__ 、 __isub__ 、 __imul__ 、 __itruediv__ 、 __ifloordiv__ 、 __imod__ 、
__ipow__
位运算符
__invert__ ~ 、 __lshift__ << 、 __rshift__ >> 、 __and__ & 、 __or__ | 、 __
xor__ ^
反向位运算符
__rlshift__ 、 __rrshift__ 、 __rand__ 、 __rxor__ 、 __ror__
增量赋值位运算符
__ilshift__ 、 __irshift__ 、 __iand__ 、 __ixor__ 、 __ior__

三、深入类和对象

鸭子类型：当看到一只鸟走起来像鸭子、游泳起来像鸭子、叫起来也像鸭子，那么这只鸟就可以被称为鸭子。

这种类型不是 tuple, list，而是iterable, callable这些

# 检查某个类是否有某种方法
hasattr(obj, "__len__")

# 判定某个对象的类型
from collections.abc import Sized
isinstance(obj, Sized)

# 强制某个子类必须实现某些方法
# 如 Web 框架需集成 cache(redis, cache, memorycache)
import abc

class CacheBase(metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def get(self,key):
        pass

    @abc.abstractmethod
    def set(self, key, value):
        pass

class RedisCache(CacheBase):
    pass

redis_cache = RedisCache()

# 执行时实现部分即会报如下错误
TypeError: Can't instantiate abstract class RedisCache with abstract methods get, set

# 检查某个类是否有某种方法

hasattr(obj, "__len__")

# 判定某个对象的类型

from collections.abc import Sized

isinstance(obj, Sized)

# 强制某个子类必须实现某些方法

# 如 Web 框架需集成 cache(redis, cache, memorycache)

import abc

class CacheBase(metaclass=abc.ABCMeta):

@abc.abstractmethod

def get(self,key):

pass

@abc.abstractmethod

def set(self, key, value):

pass

class RedisCache(CacheBase):

pass

redis_cache = RedisCache()

# 执行时实现部分即会报如下错误

TypeError: Can't instantiate abstract class RedisCache with abstract methods get, set

isinstance 和 type 的区别（推荐使用isinstance 来判断对象）

class A:
    pass

class B(A):
    pass

b = B()

print(isinstance(b, B)) # True
print(isinstance(b, A)) # True
print(type(b) is B) #True
print(type(b) is A) # False

class A:

pass

class B(A):

pass

b = B()

print(isinstance(b, B)) # True

print(isinstance(b, A)) # True

print(type(b) is B) #True

print(type(b) is A) # False

MRO（Method Resolution Order）

Python2.2以前的版本：经典类（classic class） DFS（深度优先搜索）

Python2.2版本起有了新式类（继承 object），使用 BFS（广度优先搜索），经典类仍使用 DFS

Python 3: C3算法

print(Obj.__mro__) 可打印出顺序

实例方法、静态方法（@staticmethod）、类方法（@classmethod）

私有属性：__name，但并不绝对安全的，可以通过 obj._Classname__attr 的方式调用

Python 的自省机制：自省是通过一定的机制查询到对象的内部结构（obj.__dict__, ClassName.__dict__, dir(obj), dir(ClassName))

super 函数（子类中执行super().__init__()），注意 super 函数不一定调用的是父类，而是依据前述 MRO的顺序来执行

Mixin 模式特点
1、 Mixin功能单一
2、不和基类关联，可以和任意基类组合
3、在 Mixin 中不要使用 super

上下文管理器协议（with 语句）

__enter__, __exit__

简化方式 contextlib

# 上下文管理器示例一
class Sample():
    def __enter__(self):
        print("enter")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("exit")

    def do_somthing(self):
        print("doing something")

with Sample() as sample:
    sample.do_somthing()

# 上下文管理器示例二
import contextlib

@contextlib.contextmanager
def file_open(file_name):
    print("file open")  # __enter__
    yield {}
    print("file closed") # __exit__


with file_open("test.txt") as f_opened:
    print("file processing")

# 上下文管理器示例一

class Sample():

def __enter__(self):

print("enter")

return self

def __exit__(self, exc_type, exc_val, exc_tb):

print("exit")

def do_somthing(self):

print("doing something")

with Sample() as sample:

sample.do_somthing()

# 上下文管理器示例二

import contextlib

@contextlib.contextmanager

def file_open(file_name):

print("file open") # __enter__

yield {}

print("file closed") # __exit__

with file_open("test.txt") as f_opened:

print("file processing")

四、自定义序列类

按从不同维度可分为容器序列、扁平序列和可变序列、不可变序列

容器序列（可放置不同数据类型）：list、tuple、deque

扁平序列：str、bytes、bytearray、array.array

可变序列：list、 deque、bytearray、array

不可变序列：str、tuple、bytes

+, +=和 extend 的区别：

+只能接受相同的序列类型

+=右侧可以接受任意序列类型，通过__iadd__魔法函数实现，该方法内部使用 extend 来实现

append 会将传入的序列作为一个值而非遍历地加入原序列

切片

知识回顾

#模式[start:end:step]
"""
    其中，第一个数字start表示切片开始位置，默认为0；
    第二个数字end表示切片截止（但不包含）位置（默认为列表长度）；
    第三个数字step表示切片的步长（默认为1）。
    当start为0时可以省略，当end为列表长度时可以省略，
    当step为1时可以省略，并且省略步长时可以同时省略最后一个冒号。
    另外，当step为负整数时，表示反向切片，这时start应该比end的值要大才行。
"""
aList = [3, 4, 5, 6, 7, 9, 11, 13, 15, 17]
aList[::]  # 返回包含原列表中所有元素的新列表
aList[::-1]  # 返回包含原列表中所有元素的逆序列表
aList[::2]  # 隔一个取一个，获取偶数位置的元素
aList[1::2]  # 隔一个取一个，获取奇数位置的元素
aList[3:6]  # 指定切片的开始和结束位置
aList[0:100]  # 切片结束位置大于列表长度时，从列表尾部截断
aList[100:]  # 切片开始位置大于列表长度时，返回空列表

aList[len(aList):] = [9]  # 在列表尾部增加元素
aList[:0] = [1, 2]  # 在列表头部插入元素
aList[3:3] = [4]  # 在列表中间位置插入元素
aList[:3] = [1, 2]  # 替换列表元素，等号两边的列表长度相等
aList[3:] = [4, 5, 6]  # 等号两边的列表长度也可以不相等
aList[::2] = [0] * 3  # 隔一个修改一个
print (aList)
aList[::2] = ['a', 'b', 'c']  # 隔一个修改一个
aList[::2] = [1,2]  # 左侧切片不连续，等号两边列表长度必须相等
aList[:3] = []  # 删除列表中前3个元素

del aList[:3]  # 切片元素连续
del aList[::2]  # 切片元素不连续，隔一个删一个

#模式[start:end:step]

"""

其中，第一个数字start表示切片开始位置，默认为0；

第二个数字end表示切片截止（但不包含）位置（默认为列表长度）；

第三个数字step表示切片的步长（默认为1）。

当start为0时可以省略，当end为列表长度时可以省略，

当step为1时可以省略，并且省略步长时可以同时省略最后一个冒号。

另外，当step为负整数时，表示反向切片，这时start应该比end的值要大才行。

"""

aList = [3, 4, 5, 6, 7, 9, 11, 13, 15, 17]

aList[::] # 返回包含原列表中所有元素的新列表

aList[::-1] # 返回包含原列表中所有元素的逆序列表

aList[::2] # 隔一个取一个，获取偶数位置的元素

aList[1::2] # 隔一个取一个，获取奇数位置的元素

aList[3:6] # 指定切片的开始和结束位置

aList[0:100] # 切片结束位置大于列表长度时，从列表尾部截断

aList[100:] # 切片开始位置大于列表长度时，返回空列表

aList[len(aList):] = [9] # 在列表尾部增加元素

aList[:0] = [1, 2] # 在列表头部插入元素

aList[3:3] = [4] # 在列表中间位置插入元素

aList[:3] = [1, 2] # 替换列表元素，等号两边的列表长度相等

aList[3:] = [4, 5, 6] # 等号两边的列表长度也可以不相等

aList[::2] = [0] * 3 # 隔一个修改一个

print (aList)

aList[::2] = ['a', 'b', 'c'] # 隔一个修改一个

aList[::2] = [1,2] # 左侧切片不连续，等号两边列表长度必须相等

aList[:3] = [] # 删除列表中前3个元素

del aList[:3] # 切片元素连续

del aList[::2] # 切片元素不连续，隔一个删一个

通过实现__getitem__等魔法函数即可为自定义序列添加切片等功能，示例如下：

import numbers

class Group:
    def __init__(self, group_name, company_name, staffs):
        self.group_name = group_name
        self.company_name = company_name
        self.staffs = staffs

    def __reversed__(self):
        self.staffs.reverse()

    def __getitem__(self, item):
        cls = type(self)
        if isinstance(item, slice):
            return cls(group_name=self.group_name, company_name=self.company_name, staffs=self.staffs[item])
        elif isinstance(item, numbers.Integral):
            return cls(group_name=self.group_name, company_name=self.company_name, staffs=[self.staffs[item]])

    def __len__(self):
        return len(self.staffs)

    def __iter__(self):
        return iter(self.staffs)

    def __contains__(self, item):
        if item in self.staffs:
            return True
        else:
            return False

import numbers

class Group:

def __init__(self, group_name, company_name, staffs):

self.group_name = group_name

self.company_name = company_name

self.staffs = staffs

def __reversed__(self):

self.staffs.reverse()

def __getitem__(self, item):

cls = type(self)

if isinstance(item, slice):

return cls(group_name=self.group_name, company_name=self.company_name, staffs=self.staffs[item])

elif isinstance(item, numbers.Integral):

return cls(group_name=self.group_name, company_name=self.company_name, staffs=[self.staffs[item]])

def __len__(self):

return len(self.staffs)

def __iter__(self):

return iter(self.staffs)

def __contains__(self, item):

if item in self.staffs:

return True

else:

return False

使用 bisect 来维护排序序列效率是非常高的

列表使用list 非常普遍，但有时 array（只能存放指定类型）, deque 会更好

array.array(“i”) array 的类型参见：https://docs.python.org/3/library/array.html

列表推导式、生成器表达式、字典推导式

# 列表生成式（列表推导式）
# 1、提取出1-20之间的奇数
# odd_list = []
odd_list = [i for i in range(21) if i % 2 == 1]

# 2、逻辑复杂的情况
def handle_item(item):
    return item * item
odd_list = [handle_item(i) for i in range(21) if i % 2 == 1]

# 列表生成式性能高于列表操作


# 生成器表达式
odd_gen = (i for i in range(21) if i % 2 == 1)
# generator 可通过遍历打印元素，也可转化成 List 进行操作
odd_list = list(odd_gen)

# 字典推导式
my_dict = {"demo1": 10, "demo2":20, "demo3":30}
reversed_dict = {value:key for key,value in my_dict.items()}

# 集合推导式
my_set = {key for key,value in my_dict.items()}

# 列表生成式（列表推导式）

# 1、提取出1-20之间的奇数

# odd_list = []

odd_list = [i for i in range(21) if i % 2 == 1]

# 2、逻辑复杂的情况

def handle_item(item):

return item * item

odd_list = [handle_item(i) for i in range(21) if i % 2 == 1]

# 列表生成式性能高于列表操作

# 生成器表达式

odd_gen = (i for i in range(21) if i % 2 == 1)

# generator 可通过遍历打印元素，也可转化成 List 进行操作

odd_list = list(odd_gen)

# 字典推导式

my_dict = {"demo1": 10, "demo2":20, "demo3":30}

reversed_dict = {value:key for key,value in my_dict.items()}

# 集合推导式

my_set = {key for key,value in my_dict.items()}

五、深入python的set和dict

dict属于 MutableMapping 类型，MutableMapping继承 Mapping，Mapping 继承 Collection

# 浅拷贝
my_dict.copy()
# 深拷贝
import copy
copy.deepcopy(my_dict)

# 浅拷贝

my_dict.copy()

# 深拷贝

import copy

copy.deepcopy(my_dict)

浅拷贝 dict 内部的 dict 或 list 元素在拷贝后只是一个指向，因而在新的 dict 内进行修改也会修改原 dict，而深拷贝则是一份彻底的拷贝。此外，不建议继承 List 和 Dict，因其均使用 C语言书写，正常的继承不会生效，如确需操作，可能过继承 collections.UserDict等方式。

set(集合)、frozenset(不可变集合)：无序、不重复，frozenset 可以作为 dict 的 key

set的性能很高，set 添加数据可以通过 add 方法

dict查找的性能远远大于list，list 的查找时间随着数据增大而增大，而 dict 的查找时间则几乎没有变化

1. dict的key或者set的值都必须是可以hash的
不可变对象都是可hash的， str， fronzenset， tuple，自己实现的类 __hash__
2. dict的内存花销大，但是查询速度快，自定义的对象或者python内部的对象都是用dict包装的
3. dict的存储顺序和元素添加顺序有关
4. 添加数据有可能改变已有数据的顺序

六、对象引用、可变性和垃圾回收

Python 和 Java 中的变量本质不一样，Python 的变量实质上是一个指针

is 判断两个变量的 id/内存地址是否相同，==判断两个变量值是否相等，但需要注意的是小整数和短字符串由于 Python 的 intern 机制 id 值也相同

a ="abc"
b = "abc"
print(id(a), id(b)) # 输出结果4420830968 4420830968

a ="abc"

b = "abc"

print(id(a), id(b)) # 输出结果4420830968 4420830968

Python 中垃圾回收的算法采用引用计数，只有在计数器等于0时才会回收对象

经典错误

class Company:
    def __init__(self, name, staffs=[]):
        self.name = name
        self.staffs = staffs

    def add(self, staff_name):
        self.staffs.append(staff_name)
        
    def remove(self, staff_name):
        self.staffs.remove(staff_name)

if __name__ == "__main__":
    com2 = Company("com2")
    com2.add("demo")
    print(com2.staffs) # 此处返回结果正常：['demo'] 

    com3 = Company("com3")
    com3.add("demo5")
    print(com2.staffs) # 此处结果竟与 com3相同：['demo', 'demo5']
    print(com3.staffs) # ['demo', 'demo5']
    # 原因在于这两个对象初始化时未传入 staffs列表值，因而使用了默认列表，即：
    print(Company.__init__.__defaults__) # (['demo', 'demo5'],)
    # 结论：尽量避免在方法声明中可变参数的传递，如需使用应清楚以上问题

class Company:

def __init__(self, name, staffs=[]):

self.name = name

self.staffs = staffs

def add(self, staff_name):

self.staffs.append(staff_name)

def remove(self, staff_name):

self.staffs.remove(staff_name)

if __name__ == "__main__":

com2 = Company("com2")

com2.add("demo")

print(com2.staffs) # 此处返回结果正常：['demo']

com3 = Company("com3")

com3.add("demo5")

print(com2.staffs) # 此处结果竟与 com3相同：['demo', 'demo5']

print(com3.staffs) # ['demo', 'demo5']

# 原因在于这两个对象初始化时未传入 staffs列表值，因而使用了默认列表，即：

print(Company.__init__.__defaults__) # (['demo', 'demo5'],)

# 结论：尽量避免在方法声明中可变参数的传递，如需使用应清楚以上问题

七、元类编程

通过@property可以以属性的方式来调用方法，如果要用属性的方法来设定值，则通过@xxx.setter 装饰器来定义方法

__getattr__在查找不到属性时调用, __getattribute__访问任何属性都会调用

属性描述符

实现__get__, __set__, __delete__任意方法称为属性描述符，实现了__get__和__set__的称为数据属性描述符，仅实现了__get__方法的称为非数据属性描述符

import numbers


class IntField:
    def __get__(self, instance, owner):
        return self.value

    # def __set__(self, instance, value):
    #     if not isinstance(value, numbers.Integral):
    #         raise ValueError("int value required")
    #     if value < 0:
    #         raise ValueError("positive number required")
    #     self.value = value
    #
    # def __delete__(self, instance):
    #     pass


class User:
    age = IntField()

if __name__ == "__main__":
    user = User()
    # user.age = 20
    user.__dict__["age"] = 20
    print(user.__dict__)
    print(getattr(user, 'age'))
    print(user.age)

import numbers

class IntField:

def __get__(self, instance, owner):

return self.value

# def __set__(self, instance, value):

# if not isinstance(value, numbers.Integral):

# raise ValueError("int value required")

# if value < 0:

# raise ValueError("positive number required")

# self.value = value

# def __delete__(self, instance):

# pass

class User:

age = IntField()

if __name__ == "__main__":

user = User()

# user.age = 20

user.__dict__["age"] = 20

print(user.__dict__)

print(getattr(user, 'age'))

print(user.age)

如果user是某个类的实例，那么user.age（以及等价的getattr(user,’age’)）首先调用__getattribute__，如果类定义了__getattr__方法，那么在__getattribute__抛出 AttributeError 的时候就会调用到__getattr__，而对于描述符(__get__）的调用，则是发生在__getattribute__内部的。
user = User(), 那么user.age 顺序如下：

（1）如果“age”是出现在User或其基类的__dict__中，且age是data descriptor，那么调用其__get__方法, 否则

（2）如果“age”出现在user的__dict__中，那么直接返回 obj.__dict__[‘age’]，否则

（3）如果“age”出现在User或其基类的__dict__中

（3.1）如果age是non-data descriptor，那么调用其__get__方法，否则

（3.2）返回 __dict__[‘age’]

（4）如果User有__getattr__方法，调用__getattr__方法，否则

（5）抛出AttributeError

new和init的区别

__new__用来控制对象的生成过程，在对象生成之前执行，__init__用来完善对象；如果__new__方法不返回对象（return super().__new__(cls)），则不会调用__init__方法

元类 Meta Class

# 元类是创建类的类， type->class（对象）->对象

class MetaClass(type): # 创建元类，需继承 type
    def __new__(cls, *args, **kwargs):
        return super().__new__(cls, *args, **kwargs)

class User(metaclass=MetaClass): # 使用 metaclass 来指定元类
    def __init__(self, name):
        self.name = name
    def __str__(self):
        return self.name

if __name__ == "__main__":
    my_obj = User(name="test")
    print(my_obj) # test

# 元类是创建类的类， type->class（对象）->对象

class MetaClass(type): # 创建元类，需继承 type

def __new__(cls, *args, **kwargs):

return super().__new__(cls, *args, **kwargs)

class User(metaclass=MetaClass): # 使用 metaclass 来指定元类

def __init__(self, name):

self.name = name

def __str__(self):

return self.name

if __name__ == "__main__":

my_obj = User(name="test")

print(my_obj) # test

八、迭代器和生成器

迭化器是访问集合内元素的一种方式，一般用来遍历数据。
迭代器和以下标的访问方式不一样，迭代器是不能返回的，迭代器提供了一种惰性方式的数据访问。列表等可迭代数据类型通过 __iter__来实现迭代。

获取字节码对象：dis.dis(func)

Python 生成器原理

生成器对象结构

九、Python Socket编程

应用层和传输层之前可以有一个 Socket 接口

五层网络模型

扩展阅读：TCP/IP 详解卷1-卷3

Socket 编程

Socket 编程[/caption]

# Client
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('127.0.0.1', 8000))
while True:
    re_data = input()
    client.send(re_data.encode("utf8"))
    data = client.recv(1024)
    print(data.decode("utf8"))


# Server

import socket
import threading

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8000))
server.listen()

def handle_sock(sock, addr):
    while True:
        data = sock.recv(1024)
        print(data.decode("utf8"))
        re_data = input()
        sock.send(re_data.encode("utf8"))

# 获取从客户端发送的数据，1次获取1K的数据
while True:
    sock, addr = server.accept()

    # 用线程去处理新接收的连接（用户）
    client_thread = threading.Thread(target=handle_sock, args=(sock, addr))
    client_thread.start()

# Client

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

client.connect(('127.0.0.1', 8000))

while True:

re_data = input()

client.send(re_data.encode("utf8"))

data = client.recv(1024)

print(data.decode("utf8"))

# Server

import socket

import threading

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

server.bind(('0.0.0.0', 8000))

server.listen()

def handle_sock(sock, addr):

while True:

data = sock.recv(1024)

print(data.decode("utf8"))

re_data = input()

sock.send(re_data.encode("utf8"))

# 获取从客户端发送的数据，1次获取1K的数据

while True:

sock, addr = server.accept()

# 用线程去处理新接收的连接（用户）

client_thread = threading.Thread(target=handle_sock, args=(sock, addr))

client_thread.start()

十、多线程、多进程和线程池编程

GIL: Global Interpreter Lock
Python 中一个线程对应于 C 语言中的一个线程, GIL使得同一时刻只有一个线程在一个 CPU 上执行字节码，无法将多个线程映射到多个 CPU 上执行。GIL 会根据执行的字节码行数以及时间片释放 GIL，另外GIL 在遇到 io 操作的时候也会主动释放。

from threading import Lock, RLock
# 1、锁的最大问题是会影响性能
# 2、锁会引起死锁（ acquire 前没有 release 或相互争夺资源的情况下）
# 3、使用 Rlock在同一个线程里面，可以连续调用多次 acquired，但要注意 acquire的次数要和 release 相同
lock = Lock()
lock.acquire()  # 获取锁
...             # 代码段
lock.release()  # 释放锁


############################
from threading import Condition
# Condition 的启动顺序很重要
cond = threading.Condition()
with cond:
    self.cond.wait()
    self.cond.notify()

from threading import Semaphore
# Semaphore 可用于控制指定数量的线程并发，同样拥有 acquire()和 release()方法

############################
from concurrent.futures import ThreadPoolExecutor, as_completed, wait, FIRST_COMPLETED
# 线程池：主线程中可以获取某一个线程的状态或者某一个任务的状态以及返回值。当一个线程完成的时候主线程能立即知道，futures 可以让多线程编码接口一致
executor = ThreadPoolExecutor(max_workers=2)
# 获取已经成功的 Task 返回
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=FIRST_COMPLETED)

# for future in as_completed(all_task):
#     data = future.result()
#     print("get {} page".format(data))

# 通过 executor 的 map获取已经完成的 task的值
# for data in executor.map(get_html, urls):
#     print("get {} page".format(data))

# # done 用于判定某个任务是否完成
# print(task1.done())
# # result 可以获取执行结果，是个阻塞的方法
# print(task1.result())

from threading import Lock, RLock

# 1、锁的最大问题是会影响性能

# 2、锁会引起死锁（ acquire 前没有 release 或相互争夺资源的情况下）

# 3、使用 Rlock在同一个线程里面，可以连续调用多次 acquired，但要注意 acquire的次数要和 release 相同

lock = Lock()

lock.acquire() # 获取锁

... # 代码段

lock.release() # 释放锁

############################

from threading import Condition

# Condition 的启动顺序很重要

cond = threading.Condition()

with cond:

self.cond.wait()

self.cond.notify()

from threading import Semaphore

# Semaphore 可用于控制指定数量的线程并发，同样拥有 acquire()和 release()方法

############################

from concurrent.futures import ThreadPoolExecutor, as_completed, wait, FIRST_COMPLETED

# 线程池：主线程中可以获取某一个线程的状态或者某一个任务的状态以及返回值。当一个线程完成的时候主线程能立即知道，futures 可以让多线程编码接口一致

executor = ThreadPoolExecutor(max_workers=2)

# 获取已经成功的 Task 返回

all_task = [executor.submit(get_html, (url)) for url in urls]

wait(all_task, return_when=FIRST_COMPLETED)

# for future in as_completed(all_task):

# data = future.result()

# print("get {} page".format(data))

# 通过 executor 的 map获取已经完成的 task的值

# for data in executor.map(get_html, urls):

# print("get {} page".format(data))

# # done 用于判定某个任务是否完成

# print(task1.done())

# # result 可以获取执行结果，是个阻塞的方法

# print(task1.result())

多进程编程

耗 CPU 的操作(数学计算、图像处理)，用多进程操作，对于 IO 操作使用多线程编程，因为就操作系统而言进程切换代价要高于线程。

共享全局变量不适用于多进程编程，进程间数据是隔离的，但可以通过实例化 Manager().dict()在实现该数据共享。

multiprocessing 中的 Queue 不能用于 pool 进程池，pool 中进程间通信需要使用 Manager 中的 Queue

通过 Pipe 实现进程间通信，Pipe 性能高于 Queue，但Pipe 只能适用于两个进程

十一、协程和异步io

并发是指一个时间段内，有几个程序在同一个 CPU 上运行，但任意时刻只有一个程序在 CPU 上运行

并行是批在任意时刻点上，有多个程序同时运行在多个CPU上

同步是指代码调用 IO 操作时，必须等待 IO 操作完成才返回的调用方式

异步是指代码调用 IO 操作时，不必等待 IO 操作完成就返回的调用方式

阻塞是指调用函数时当前线程被挂起

非阻塞是指调用函数时当前线程不会被挂起，而是立即返回

Unix 下的五种I/O模型：阻塞式 I/O、非阻塞式 I/O、I/O复用、信号驱动式I/O、异步I/O

select, poll, epoll

生成器

生成器不止可以产出值，还可以接收值

在调用 send 发送非 None 值之前，必须启动一次生成器，方式有两种：

1、gen.send(None)，2、next(gen)

send 方法可以传递值到生成器内部，还可以重启生成器到下一个位置

GeneratorExit继承自 BaseException，而非 Exception

import types
# from collections.abc import Awaitable
# Python 为了将语义变得更加明确，就引入了 async 和 await 关键词用于定义原生的协程
# await 后面需要跟 Awaitable 对象
@types.coroutine
def downloader(url):
    yield "demo"

# async def downloader(url):
#     return "demo"

async def download_url(url):
     html = await downloader(url)
     return html

if __name__ == "__main__":
    coro = download_url("https://www.baidu.com")
    coro.send(None)

import types

# from collections.abc import Awaitable

# Python 为了将语义变得更加明确，就引入了 async 和 await 关键词用于定义原生的协程

# await 后面需要跟 Awaitable 对象

@types.coroutine

def downloader(url):

yield "demo"

# async def downloader(url):

# return "demo"

async def download_url(url):

html = await downloader(url)

return html

if __name__ == "__main__":

coro = download_url("https://www.baidu.com")

coro.send(None)

十二、asyncio并发编程

asyncio

包含各种特定系统实现的模块化事件循环
传输和协议抽象
对 TCP、UDP、SSL、子进程、延时调用以及其它的具体支持
模仿 futures 模块但适用于事件循环使用的 Future 类
基于 yield from 的协议和任务，可以让你用顺序的方式编写并发代码
必须使用一个将产生阻塞 IO 的调用时，有接口可以把这个事件转移到线程池
模仿 threading 模块中的同步原语，可以用在单线程内的协程之间

# asyncio 是 Python 用于解决异步 IO 编程的一整套解决方案   
import asyncio           
from functools import partial  

async def get_html(url):
    print("start get url")
    await asyncio.sleep(2)
    print("end get url")
    # return "demo"

if __name__ == "__main__":
    tasks = [get_html("https://www.baidu.com") for i in range(10)]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    # get_future = asyncio.ensure_future(get_html("https://www.baidu.com")) # loop.create_task()几乎等价
    # task = loop.create_task(get_html("https://www.baidu.com"))
    # task.add_done_callback(partial(callback, "https://www.baidu.com"))  
    # loop.run_until_complete(get_html("https://www.baidu.com"))
    # print(task.result()) # 打印返回值
    # loop.run_until_complete(asyncio.gather(*tasks))   # gather较wait更加 high-level  
    # call_later, call_at, call_soon, call_soon_threadsafe

# asyncio 是 Python 用于解决异步 IO 编程的一整套解决方案

import asyncio

from functools import partial

async def get_html(url):

print("start get url")

await asyncio.sleep(2)

print("end get url")

# return "demo"

if __name__ == "__main__":

tasks = [get_html("https://www.baidu.com") for i in range(10)]

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

# get_future = asyncio.ensure_future(get_html("https://www.baidu.com")) # loop.create_task()几乎等价

# task = loop.create_task(get_html("https://www.baidu.com"))

# task.add_done_callback(partial(callback, "https://www.baidu.com"))

# loop.run_until_complete(get_html("https://www.baidu.com"))

# print(task.result()) # 打印返回值

# loop.run_until_complete(asyncio.gather(*tasks)) # gather较wait更加 high-level

# call_later, call_at, call_soon, call_soon_threadsafe

aiohttp实现高并发爬虫

import asyncio
import re

import aiohttp
import aiomysql
from pyquery import PyQuery

stopping = False
start_url = "http://www.jobbole.com/"
waiting_urls = []
seen_urls = set()

sem = asyncio.Semaphore(3) # 并发控制

async def fetch(url, session):
    async with sem:
        # await asyncio.sleep(1)
        try:
            async with session.get(url) as resp:
                print("url status:{}".format(resp.status))
                if resp.status in [200,201]:
                    data = await resp.text()
                    return data
        except Exception as e:
            print(e)
def extract_urls(html):
    urls = []
    pq = PyQuery(html)
    for link in pq.items("a"):
        url = link.attr("href")
        if url and url.startswith("http") and url not in seen_urls:
            urls.append(url)
            waiting_urls.append(url)
    return urls

async def init_urls(url, session):
    html = await fetch(url, session)
    seen_urls.add(url)
    extract_urls(html)

async def article_handler(url, session, pool):
    # 获取文章详情并解析入库，测试仅放置一个字段: title
    html = await fetch(url, session)
    seen_urls.add(url)
    extract_urls(html)
    pq = PyQuery(html)
    title = pq("title").text()
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            insert_sql = "INSERT INTO article(title) VALUES ('{}')".format(title)
            await cur.execute(insert_sql)


async def consumer(pool):
    async with aiohttp.ClientSession() as session:
        while not stopping:
            if len(waiting_urls) == 0:
                await asyncio.sleep(0.5)
                continue
            url = waiting_urls.pop()
            print("start get url:{}".format(url))
            if re.match('http://.*?jobbole.com/\d+/', url):
                if url not in seen_urls:
                    asyncio.ensure_future(article_handler(url, session, pool))
                    # await asyncio.sleep(0.5)
            else:
                if url not in seen_urls:
                    asyncio.ensure_future(init_urls(url, session))


async def main(loop):
    # 等待 MySQL连接
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
                                      user='root', password='',
                                      db='aiotest', loop=loop, charset="utf8", autocommit=True)

    async with aiohttp.ClientSession() as session:
        html = await fetch(start_url, session)
        seen_urls.add(start_url)
        extract_urls(html)

    asyncio.ensure_future(consumer(pool))


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(main(loop))
    loop.run_forever()

import asyncio

import re

import aiohttp

import aiomysql

from pyquery import PyQuery

stopping = False

start_url = "http://www.jobbole.com/"

waiting_urls = []

seen_urls = set()

sem = asyncio.Semaphore(3) # 并发控制

async def fetch(url, session):

async with sem:

# await asyncio.sleep(1)

try:

async with session.get(url) as resp:

print("url status:{}".format(resp.status))

if resp.status in [200,201]:

data = await resp.text()

return data

except Exception as e:

print(e)

def extract_urls(html):

urls = []

pq = PyQuery(html)

for link in pq.items("a"):

url = link.attr("href")

if url and url.startswith("http") and url not in seen_urls:

urls.append(url)

waiting_urls.append(url)

return urls

async def init_urls(url, session):

html = await fetch(url, session)

seen_urls.add(url)

extract_urls(html)

async def article_handler(url, session, pool):

# 获取文章详情并解析入库，测试仅放置一个字段: title

html = await fetch(url, session)

seen_urls.add(url)

extract_urls(html)

pq = PyQuery(html)

title = pq("title").text()

async with pool.acquire() as conn:

async with conn.cursor() as cur:

insert_sql = "INSERT INTO article(title) VALUES ('{}')".format(title)

await cur.execute(insert_sql)

async def consumer(pool):

async with aiohttp.ClientSession() as session:

while not stopping:

if len(waiting_urls) == 0:

await asyncio.sleep(0.5)

continue

url = waiting_urls.pop()

print("start get url:{}".format(url))

if re.match('http://.*?jobbole.com/\d+/', url):

if url not in seen_urls:

asyncio.ensure_future(article_handler(url, session, pool))

# await asyncio.sleep(0.5)

else:

if url not in seen_urls:

asyncio.ensure_future(init_urls(url, session))

async def main(loop):

# 等待 MySQL连接

pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,

user='root', password='',

db='aiotest', loop=loop, charset="utf8", autocommit=True)

async with aiohttp.ClientSession() as session:

html = await fetch(start_url, session)

seen_urls.add(start_url)

extract_urls(html)

asyncio.ensure_future(consumer(pool))

if __name__ == "__main__":

loop = asyncio.get_event_loop()

asyncio.ensure_future(main(loop))

loop.run_forever()

一、Python中一切皆对象

type、object和class的关系

Python中的内置类型

二、魔法函数

非数学运算

数学运算

三、深入类和对象

四、自定义序列类

五、深入python的set和dict

六、对象引用、可变性和垃圾回收

七、元类编程

属性描述符

__new__和__init__的区别

元类 Meta Class

八、迭代器和生成器

九、Python Socket编程

十、多线程、多进程和线程池编程

多进程编程

十一、协程和异步io

生成器

十二、asyncio并发编程

aiohttp实现高并发爬虫

相关资源：

new和init的区别