python总结

Python语言基础

python语言特性

• 动态强类型语言(不少人误以为是弱类型)
• 动态还是静态指的是编译期还是运行期确定类型
• 强类型指的是不会发生隐式类型转换

鸭子类型

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

• 关注点在对象的行为，而不是类型(duck typing)
• 比如 file,StringIO,socket 对象都支持read/write方法(file like object)
• 再比如定义了__iter__魔术方法的对象可以用 for 迭代

python2与3的差异

• python3 中print 成为函数
• 编码问题，python3中str默认就是unicode
• Python3 除号返回浮点数
• 优化的 super() 方便直接调用父类函数
• python3新增的：
  • yield from 链接子生成器
  • asyncio内置库，async/await 原生协程支持异步编程
• python3改进的：
  • 一些内置库的修改。urllib, selector 等
  • 生成的 pyc 文件统一放到 __pycache__

monkey patch

• 所谓的 monkey patch 就是运行时替换
• 比如 gevent 库需要修改内置的 socket
• from gevent import monkey; monkey.patch_socket()

自省Introspection

• 运行时判断一个对象的类型的能力
• Python一切皆对象，用 type, id, isinstance 获取对象类型信息
• Inspect 模块提供了更多获取对象信息的函数

python函数

可变类型作为参数：

def flist(lt):
    lt.append(0)
    print(lt)
    
lt = []
flist(lt)
flist(lt)

# 输出结果：
[0]
[0, 0]

不可变类型作为参数：

def fstr(s):
    s += "a"
    print(s)
    
s = "hehe"
fstr(s)
fstr(s)

# 输出结果：
hehea
hehea

python传递参数：唯一支持的参数传递是共享传参，函数形参获得实参中各个引用的副本

def clear_list(lt):
    lt = []
    
ll = [1, 2, 3]
clear_list(ll)
print(ll)

# 输出结果：
[1, 2, 3]

python可变参数作为默认参数:

默认参数只计算一次

def func(lt=[1]):
    lt.append(2)
    print(lt)

func()
func()
# 输出结果：
[1, 2]
[1, 2, 2]

python异常处理

python异常类

• BaseException， 最基础的
• SystemExit/KeyboardInterrupt/GeneratorExit
• Exception 自定义异常类，一般继承这个

try:
    func()      # 可能会抛出异常代码
except (Exception1, Exception2) as e:   # 捕获多个异常并处理
    # 异常处理的代码
else:
    # 异常没有发生时的代码逻辑
finally:
    # 无论异常有没有发生都会执行的代码。一般处理资源的释放和关闭

python性能

GIL的影响

• 限制了程序的多核执行
• 同一个时间只能有一个线程执行字节码
• CPU密集程序难以利用多核优势
• IO期间会释放 GIL，对IO密集程序影响不大

为什么有了GIL还要关注线程安全：

• 一个操作如果是一个字节码指令可以完成就是原子的
• 原子的是可以保证线程安全的
• 使用 dis 操作来分析字节码

dis分析原子操作:

import dis
def update_list(lt):
    lt[0] = 1
    
dis.dis(update_list)

"""
  2           0 LOAD_CONST               1 (1)
              2 LOAD_FAST                0 (lt)
              4 LOAD_CONST               2 (0)
              6 STORE_SUBSCR
              8 LOAD_CONST               0 (None)
             10 RETURN_VALUE
"""

dis分析非原子操作：

import dis

def incr_list(lt):
    lt[0] += 1   # 不是原子操作
    
dis.dis(incr_list)

"""
  2           0 LOAD_FAST                0 (lt)
              2 LOAD_CONST               1 (0)
              4 DUP_TOP_TWO
              6 BINARY_SUBSCR
              8 LOAD_CONST               2 (1)
             10 INPLACE_ADD	   # 需要多个字节码操作，有可能在执行过程中切换到其他线程
             12 ROT_THREE
             14 STORE_SUBSCR
             16 LOAD_CONST               0 (None)
             18 RETURN_VALUE
"""

剖析程序性能：

• 内置的 profile/cprofile 等工具
• 使用 pyflame(uber开源)的火焰图工具

服务端性能优化措施：

• 数据结构与算法优化
• 数据库层：索引优化，慢查询消除，批量操作减少IO，NoSQL
• 网络IO：批量操作，pipeline操作 减少IO
• 缓存
• 异步
• 并发

生成器与协程

生成器：

• 生成器就是可以生成值的函数
• 当一个函数里有了 yield 关键字就成了生成器
• 生成器可以挂起执行并且保持当前执行的状态

def simple_gen():
    yield "hello"
    yield "world"
    
gen = simple_gen()
print(next(gen))
print(next(gen))

基于生成器的协程：

Python3之前没有原生协程，只有基于生成器的协程

• pep 342(Coroutines via Enhanced Generators)增强生成器功能
• 生成器可以通过yield暂停执行和产出数据
• 同时支持send()向生成器发送数据和throw()向生成器抛异常

def coro():
    hello = yield "hello"    # yield关键字=右边作为表达式，可以被send值
    yield hello

c = coro()
print(next(c)) # 输出hello，这里调用next产出第一个值hello，之后函数暂停
print(c.send("world")) # 再次调用send发送值，此时hello变量为world，然后yield 产出hello变量的值 world
# 之后协程结束，若再send值会抛出异常StopIteration

使用基于生成器的协程 注意点：

• 协程使用需要send(None)或者next(coroutine)来预激，才能启动
• 在yield处协程会暂停执行
• 单独的yield value 会产出值给调用方
• 可以通过coroutine.send(value)来给协程发送值，发送的值会赋值给yield表达式左边的变量
• 协程执行完成后会抛出StopIteration异常

协程装饰器

# 避免每次都要用 send 预激它
from functools import wraps
def coroutine(func):
    @wraps(func)
    def primer(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)    # 预激
        return gen
    return primer

python数据结构与算法

常用的内置算法与数据结构：

import collections

Point = collections.namedtuple("Point", 'x, y')
p = Point(1, 2)
print(p.x)  # 1
print(p.y)  # 2  # namedtuple让 tuple属性变得可读

de = collections.deque()
de.append(1)
de.appendleft(0)
print(de)   # deque([0, 1])
de.pop()
de.popleft()

c = collections.Counter("abcabcc")
print(c['c'])   # 3
print(c.most_common())    # [('c', 3), ('a', 2), ('b', 2)]


od = collections.OrderedDict()
od['c'] = 'c'
od['a'] = 'a'
od['b'] = 'b'
print(list(od.keys()))      # ['c', 'a', 'b']  key的顺序


# 带有默认值的字典
dd = collections.defaultdict(int)
print(dd['a'])    # 0
dd['b'] += 1
print(dd)    # defaultdict(<class 'int'>, {'a': 0, 'b': 1})

python dict底层结构

dict底层使用哈希表

• 为了支持快速查找使用了哈希表作为底层结构
• 哈希表平均查找时间复杂度O(1)
• CPython 解释器使用二次探查解决哈希冲突问题
• 哈希扩容问题

python字典为什么这么快 https://blog.csdn.net/weixin_42681866/article/details/82785127

在Python中，字典是通过散列表（哈希表）实现的。字典也叫哈希数组或关联数组，所以其本质是数组（如下图），每个 bucket 有两部分：一个是键对象的引用，一个是值对象的引用。所有 bucket 结构和大小一致，我们可以通过偏移量来读取指定 bucket。

通过哈希函数计算键对象name的哈希值，对数组长度取余hash(hashable)%k，因为哈希值最右3位110为十进制6，则查看数组索引6对应的bucket是否为空，如果为空则将键值对放入，否则（哈希冲突）左移三位即000，查看索引0是否为空，循环直至找到空的bucket。

Python会根据哈希数组的拥挤程度对其扩容。“扩容”指的是:创造更大的数组，这时候会对已经存在的键值对重新进行哈希取余运算保存到其它位置；一般接近 2/3 时，数组就会扩容。扩容后，偏移量的数字个数增加，如数组长度扩容到16时，可以用最右边4位数字作为偏移量。

https://zhuanlan.zhihu.com/p/188640527

对于散列表而言，无论设置的存储区域（n）有多大，当需要存储的数据大于 n 时，那么必然会存在哈希值相同的情况。这就是所谓的散列冲突。

哈希冲突解决方法有两类：开放寻址法（open addressing）和链表法（chaining）

开放寻址法：

将散列函数扩展定义成探查序列，即每个关键字有一个探查序列h(k,0)、h(k,1)、…、h(k,m-1)，这个探查序列一定是0….m-1的一个排列（一定要包含散列表全部的下标，不然可能会发生虽然散列表没满，但是元素不能插入的情况），如果给定一个关键字k，首先会看h(k,0)是否为空，如果为空，则插入；如果不为空，则看h(k,1)是否为空。

开放寻址法是一种解决碰撞的方法，对于开放寻址冲突解决方法，比较经典的有线性探测方法（Linear Probing）、二次探测（Quadratic probing）和 双重散列（Double hashing）等方法

线性探测方法：

• 当我们往散列表中插入数据时，如果某个数据经过散列函数散列之后，存储位置已经被占用了，我们就从当前位置开始，依次往后查找，看是否有空闲位置，直到找到为止
• 线性探测法一个很大的弊端就是当散列表中插入的数据越来越多时，散列冲突发生的可能性就会越来越大，空闲位置会越来越少，线性探测的时间就会越来越久。极端情况下，需要从头到尾探测整个散列表，所以最坏情况下的时间复杂度为 O(n)

二次探测：

• 二次探测是二次方探测法的简称。顾名思义，使用二次探测进行探测的步长变成了原来的“二次方”，也就是说，它探测的下标序列为 hash(key)+0，hash(key)+1^2或[hash(key)-1^2]，hash(key)+2^2或[hash(key)-2^2]。

• 以上图为例，散列表的大小为 8 ，黄色区域表示空闲位置，橙色区域表示已经存储了数据。目前散列表中已经存储了 7 个元素。此时元素 7777777 经过 Hash 算法之后，被散列到位置下标为 7 的位置，但是这个位置已经有数据了，所以就产生了冲突。

  按照二次探测方法的操作，有冲突就先 + 1^2，8 这个位置有值，冲突；变为 - 1^2，6 这个位置有值，还是有冲突；于是 - 2^2， 3 这个位置是空闲的，插入

双重散列方法:

• 所谓双重散列，意思就是不仅要使用一个散列函数，而是使用一组散列函数 hash1(key)，hash2(key)，hash3(key)。。。。。。先用第一个散列函数，如果计算得到的存储位置已经被占用，再用第二个散列函数，依次类推，直到找到空闲的存储位置。

• 以上图为例，散列表的大小为 8 ，黄色区域表示空闲位置，橙色区域表示已经存储了数据。目前散列表中已经存储了 7 个元素。此时元素 7777777 经过 Hash 算法之后，被散列到位置下标为 7 的位置，但是这个位置已经有数据了，所以就产生了冲突。

  此时，再将数据进行一次哈希算法处理，经过另外的 Hash 算法之后，被散列到位置下标为 3 的位置，完成操作

事实上，不管采用哪种探测方法，只要当散列表中空闲位置不多的时候，散列冲突的概率就会大大提高。为了尽可能保证散列表的操作效率，一般情况下，需要尽可能保证散列表中有一定比例的空闲槽位。

一般使用加载因子（load factor）来表示空位的多少。

加载因子是表示 Hsah 表中元素的填满的程度，若加载因子越大，则填满的元素越多,这样的好处是：空间利用率高了,但冲突的机会加大了。反之,加载因子越小,填满的元素越少,好处是冲突的机会减小了，但空间浪费多了。

链表法：

• 链表法是一种更加常用的散列冲突解决办法，相比开放寻址法，它要简单很多。如下动图所示，在散列表中，每个位置对应一条链表，所有散列值相同的元素都放到相同位置对应的链表中。

LRU Cache

Least-Recently-Used 替换掉最近最少使用的对象

• 缓存剔除策略，当缓存空间不够用的时候需要一种方式剔除key
• 常见的有 LRU，LFU 等
• LRU通过使用一个循环双端队列不断把最新访问的key 放到表头实现

实现LRUCache

• 字典用来缓存，循环双端链表用来记录访问顺序
• 利用 Python 内置的 dict + collections.OrderedDict 实现
• dict 用来当做 k/v 键值对的缓存
• OrderedDict 用来实现更新最近访问的 key

from collections import OrderedDict


class LRUCache:
    def __init__(self, capacity=128):
        self.capacity = capacity
        self.od = OrderedDict()

    # 每次访问更新最新使用的key
    def get(self, key):
        if key in self.od:
            val = self.od[key]
            self.od.move_to_end(key)  # 把最近访问的放在尾部，就是最新的
            return val
        else:
            return -1

    # 更新 k / v
    def put(self, key, value):
        if key in self.od:
            del self.od[key]
            self.od[key] = value  # 相当于更新key到表头，表示最新访问的
        else:
            self.od[key] = value
            # 判断当前容量是否满了
            if len(self.od) > self.capacity:
                self.od.popitem(last=False)  # 容量满了删除最早的key

python常考算法

排序 + 查找

• 常考排序算法：冒泡排序、快速排序、归并排序、堆排序
• 线性查找，二分查找等

什么是排序算法的稳定性？

• 相同大小的元素在排序之后依然保持相对位置不变，就是稳定的
• r[i]=r[j] 且 r[i]在r[j] 之前，排序之后 r[i] 依然在 r[j] 之前
• 稳定性对于排序一个复杂结构，并且需要保持原有排序才有意义

快速排序

快排经常问：分治法(divide and conquer)，快排三步走：

• Partition: 选择基准分割数组为两个子数组，小于基准和大于基准的
• 对两个子数组分别快排，(递归处理)
• 合并结果

def quick_sort(array):
    # 递归出口
    if len(array) < 2:
        return array
    else:
        pivot_index = 0  # 选择第一个元素作为基准元素
        pivot = array[pivot_index]
        left_sort = [
            i for i in array[pivot_index + 1:] if i <= pivot
        ]
        right_sort = [
            i for i in array[pivot_index + 1:] if i > pivot
        ]
        return quick_sort(left_sort) + [pivot] + quick_sort(right_sort)

def test():
    import random
    lt = list(range(10))
    random.shuffle(lt)
    print(lt)
    print(quick_sort(lt))

test()

归并排序

实现合并两个有序数组

def merge_sorted_list(sorted_a, sorted_b):
    length_a, length_b = len(sorted_a), len(sorted_b)
    # 设置两个指针
    a = b = 0
    new_sorted_seq = []

    while a < length_a and b < length_b:
        if sorted_a[a] < sorted_b[b]:
            new_sorted_seq.append(sorted_a[a])
            a += 1
        else:
            new_sorted_seq.append(sorted_b[b])
            b += 1

    if a < length_a:
        new_sorted_seq.extend(sorted_a[a:])
    else:
        new_sorted_seq.extend(sorted_b[b:])

    return new_sorted_seq


def test_merge_sorted_list():
    a = [1, 2, 5]
    b = [0, 3, 4, 7]
    print(merge_sorted_list(a, b))


test_merge_sorted_list()

归并排序：

def merge_sorted_list(sorted_a, sorted_b):
    length_a, length_b = len(sorted_a), len(sorted_b)
    # 设置两个指针
    a = b = 0
    new_sorted_seq = []

    while a < length_a and b < length_b:
        if sorted_a[a] < sorted_b[b]:
            new_sorted_seq.append(sorted_a[a])
            a += 1
        else:
            new_sorted_seq.append(sorted_b[b])
            b += 1

    if a < length_a:
        new_sorted_seq.extend(sorted_a[a:])
    else:
        new_sorted_seq.extend(sorted_b[b:])

    return new_sorted_seq


def test_merge_sorted_list():
    a = [1, 2, 5]
    b = [0, 3, 4, 7]
    print(merge_sorted_list(a, b))


def mergesort(array):
    # 递归出口
    if len(array) <= 1:
        return array
    else:
        mid = int(len(array) / 2)
        left_half = mergesort(array[:mid])
        right_half = mergesort(array[mid:])
        return merge_sorted_list(left_half, right_half)


def test_mergesort():
    import random
    lt = list(range(10))
    random.shuffle(lt)
    print(lt)
    print(mergesort(lt))


test_mergesort()

堆排序

from heapq import heappush, heappop


def heapsort(iterable):
    items = []
    for value in iterable:
        heappush(items, value)
    return [heappop(items) for i in range(len(items))]

二分查找

def binary_search(sorted_array, val):
    if not sorted_array:
        return -1

    beg, end = 0, len(sorted_array) - 1

    while beg <= end:
        mid = int((beg + end) / 2)
        if sorted_array[mid] == val:
            return mid
        elif sorted_array[mid] > val:  # 要找的值小于中间的值，就往左边找
            end = mid - 1
        else:
            beg = mid + 1

    return -1

python常考数据结构

链表

单链表、双链表、循环双端链表

反转链表：

"""
Input: 1->2->3->4->5->NULL
Output:5->4->3->2->1->NULL
"""

class ListNode:
    def __init__(self, data, next=None):
        self.val = data
        self.next = next

def reverseList(head):
    pre = None
    cur = head
    # 把cur值前移，pre也前移，相当于把之前的cur值赋值给pre，一直到cur指向了None
    while cur:
        nextnode = cur.next
        cur.next = pre  # cur往回指
        pre = cur
        cur = nextnode

    return pre

if __name__ == '__main__':
    link = ListNode(1, ListNode(2, ListNode(3, ListNode(4, ListNode(5)))))
    root = reverseList(link)
    while root:
        print(root.val)
        root = root.next
        
 # 输出结果：
5
4
3
2
1

删除链表中的节点

"""
请编写一个函数，使其可以删除某个链表中给定的（非末尾）节点，你将只被给定要求被删除的节点。
示例 1:

输入: head = [4,5,1,9], node = 5
输出: [4,1,9]
解释: 给定你链表中值为 5 的第二个节点，那么在调用了你的函数之后，该链表应变为 4 -> 1 -> 9.
"""

队列

• 实现队列的 apend 和 pop 操作，如何做到先进先出
• 使用 Python 的 list 或者 collections.deque 实现队列

from collections import deque

class Queue:
    def __init__(self):
        self.items = deque()

    def append(self, val):
        return self.items.append(val)   # 右边加

    def pop(self):
        return self.items.popleft()    # 左边取

    def empty(self):
        return len(self.items) == 0

def test_queue():
    q = Queue()
    q.append(0)
    q.append(1)
    q.append(2)
    print(q.pop())
    print(q.pop())
    print(q.pop())

test_queue()

栈

实现栈的 push 和 pop 操作，后进先出

from collections import deque


class Stack:
    def __init__(self):
        self.deque = deque()

    def push(self, val):
        self.deque.append(val)

    def pop(self):
        self.deque.pop()

如何用两个栈实现一个队列：

• 将stack1作为存储空间，将stack2作为临时缓冲区；也就是stack2辅助stack1做出队与入队操作
• 入队时，直接将元素压入stack1即可
• 出队时，先判断stack2中是否有数据
  • 如果有数据的话，直接pop() stack2中的数据（因为pop是直接弹出最后一个数据）
  • 如果stack2中没有数据，则将stack1中的数据，push到stack2中，然后stack2中就有了先到的数据，实现了队列的的先进先出

"""
用两个栈实现一个队列
"""

from collections import deque


class Stack:
    def __init__(self):
        self.items = deque()

    def push(self, val):
        return self.items.append(val)

    def pop(self):
        return self.items.pop()

    def top(self):
        return self.items[-1]  # 返回栈顶值

    def empty(self):
        return len(self.items) == 0


class MyQueue:
    def __init__(self):
        self.s1 = Stack()
        self.s2 = Stack()

    def push(self, x):
        self.s1.push(x)

    def pop(self):
        if not self.s2.empty():
            return self.s2.pop()
        while not self.s1.empty():
            val = self.s1.pop()
            self.s2.push(val)
        return self.s2.pop()

    def peek(self):
        if not self.s2.empty():
            return self.s2.top()
        while not self.s1.empty():
            val = self.s1.pop()
            self.s2.push(val)
        return self.s2.top()

    def empty(self):
        return self.s1.empty() and self.s2.empty()


def test():
    q = MyQueue()
    q.push(1)
    q.push(2)
    q.push(3)
    print(q.pop())
    print(q.pop())
    print(q.pop())


test()

字典和集合

Python dict/set 底层都是哈希表

• 哈希表的实现原理，底层其实就是一个数组
• 根据哈希函数快速定位一个元素，平均查找 O(1)，非常快
• 不断加入元素会引起哈希表重新开辟空间，拷贝之前元素到新数组

哈希冲突

二叉树

先序、中序、后序遍历

• 先(根)序：先处理根，之后是左子树，然后是右子树
• 中(根)序：先处理左子树，然后是跟，然后是右子树
• 后(根)序：先处理左子树，然后是右子树，最后是根

树的遍历：

先序遍历，其实很简单，递归代码里先处理根就好了

class BinTreeNode:
    def __init__(self, data, left=None, right=None):
        self.data = data
        self.left = left
        self.right = right


class BinTree:
    def __init__(self, root=None):
        self.root = root

    def preorder_trav(self, subtree):
        "先序遍历"
        if subtree is not None:
            print(subtree.data)   # 先处理根
            self.preorder_trav(subtree.left)     # 递归处理左子树
            self.preorder_trav(subtree.right)    # 递归处理右子树

堆

堆其实是完全二叉树，有最大堆和最小堆

• 最大堆：对于每个非叶子节点V，V 的值都比它的两个孩子大
• 最大堆支持每次 pop 操作获取最大的元素，最小堆获取最小元素
• 常见问题：用堆来完成 topk 问题，从海量数字中寻找最大的 k 个

import heapq


class TopK:
    """
    获取大量元素 topK 大个元素，固定内存
    思路：
    1、先放入元素前k个建立一个最小堆
    2、迭代剩余元素：
        如果当前元素小于堆顶元素，跳过该元素
        否则替换堆顶元素为当前元素，并重新调整堆
    """

    def __init__(self, iterable, k):
        self.minheap = []
        self.iterable = iterable
        self.capacity = k

    def push(self, val):
        if len(self.minheap) >= self.capacity:
            min_val = self.minheap[0]
            if val < min_val:
                pass
            else:
                heapq.heapreplace(self.minheap, val)  # 返回并且pop 堆顶最小值，推入新的val值并调整堆

        else:
            heapq.heappush(self.minheap, val)  # 前k个元素直接放入minheap

    def get_topk(self):
        for val in self.iterable:
            self.push(val)
        return self.minheap


def test():
    import random
    lt = list(range(1000))
    random.shuffle(lt)
    aa = TopK(lt, 5)
    print(aa.get_topk())  # [995, 996, 999, 997, 998]


test()

字符串

反转字符串

def reverseStr(s):
    beg, end = 0, len(s) - 1
    while beg < end:
        s[beg], s[end] = s[end], s[beg]
        beg += 1
        end -= 1
    return s

print(reverseStr(["a", "b", "c", "d"]))

判断回文数

def isPalindrome(x):
    if x < 0:
        return False
    s = str(x)
    beg, end = 0, len(s) - 1
    while beg < end:
        if s[beg] == s[end]:
            beg += 1
            end -= 1
        else:
            return False
    return True

print(isPalindrome(1234321))

python面向对象

装饰器

# 编写一个记录函数耗时的装饰器

from functools import wraps
import time


def log_time(func):
    @wraps(func)
    def log(*args, **kwargs):
        start = time.time()
        res = func(*args, **kwargs)
        print("use time: {}".format(time.time() - start))
        return res

    return log


@log_time
def mysleep():
    time.sleep(1)


mysleep()

使用类来编写装饰器

from functools import wraps
import time


class LogTime:
    def __call__(self, func):
        @wraps(func)
        def _log(*args, **kwargs):
            start_time = time.time()
            res = func(*args, **kwargs)
            print(f"use time: {time.time() - start_time}")
            return res

        return _log


@LogTime()
def mysleep2():
    time.sleep(1)


mysleep2()

给装饰器加参数：

from functools import wraps
import time


class LogTimeParams:
    def __init__(self, use_int=False):
        self.use_int = use_int

    def __call__(self, func):
        @wraps(func)
        def _log(*args, **kwargs):
            start_time = time.time()
            res = func(*args, **kwargs)
            print(f"use time: {time.time() - start_time}")
            return res

        return _log


@LogTimeParams()
def mysleep2():
    time.sleep(1)


mysleep2()

常见的设计模式

常见创建型设计模式：

• 工厂模式(Factory): 解决对象创建问题
• 构造模式(Builder): 控制复杂对象的创建
• 原型模式(Prototype)：通过原型的克隆创建新的实例
• 单例(Borg/Singleton): 一个类只能创建同一个对象
• 对象池模式(Pool): 预先分配同一类型的一组实例
• 惰性计算模式(Lazy Evaluation)：延迟计算(python 的property)

工厂模式

• 解决对象创建问题
• 解耦对象的创建和使用
• 包括工厂方法和抽象工厂

# 一个工厂方法的例子
class DogToy:
    def speak(self):
        print("wang wang")


class CatToy:
    def speak(self):
        print("miao miao")


def toy_factory(toy_type):
    if toy_type == 'dog':
        return DogToy()
    elif toy_type == 'cat':
        return CatToy()

构造模式：

• 用来控制复杂对象的构造
• 创建和表示分离。比如你要买电脑，工厂模式直接给你需要的电脑
• 但是构造模式允许你自己定义电脑的配置，组装完成后给你

# 一个构造模式的例子
class Computer:
    def __init__(self, serial_number):
        self.serial = serial_number
        self.memory = None  # in gigabytes
        self.hdd = None  # in gigabytes
        self.gpu = None

    def __str__(self):
        info = ('Memory: {}GB'.format(self.memory),
                'Hard Disk: {}GB'.format(self.hdd),
                'Graphics Card: {}'.format(self.gpu))
        return '\n'.join(info)


class ComputerBuilder:
    def __init__(self):
        self.computer = Computer('AG23385193')

    def configure_memory(self, amount):
        self.computer.memory = amount

    def configure_hdd(self, amount):
        self.computer.hdd = amount

    def configure_gpu(self, gpu_model):
        self.computer.gpu = gpu_model


class HardwareEngineer:
    def __init__(self):
        self.builder = None

    def construct_computer(self, memory, hdd, gpu):
        self.builder = ComputerBuilder()
        var = [step for step in (self.builder.configure_memory(memory),
                                 self.builder.configure_hdd(hdd),
                                 self.builder.configure_gpu(gpu))]

    @property
    def computer(self):
        return self.builder.computer


# 使用buidler，可以创建多个builder类实现不同的组装方式
engineer = HardwareEngineer()
engineer.construct_computer(hdd=1000, memory=32, gpu='GeForce RTX 3070')
computer = engineer.computer
print(computer)

原型模式：

• 通过克隆原型来创建新的实例
• 可以使用相同的原型，通过修改部分属性来创建新的示例
• 用途：对于一些创建实例开销比较高的地方可以用原型模式

单例模式：

• 单例模式：一个类创建出来的对象都是同一个
• Python的模块其实就是单例的，只会导入一次
• 使用共享同一个实例的方式来创建单例模式

# 单例模式
class Singleton:
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, '_instance'):
            _instance = super().__new__(cls, *args, **kwargs)
            cls._instance = _instance
        return cls._instance


class MyClass(Singleton):
    pass


c1 = MyClass()
c2 = MyClass()
assert c1 is c2  # 单例的，c1 c2 同一个实例

常见结构型设计模式

• 装饰器模式(Decorator): 无需子类化扩展对象功能
• 代理模式(Proxy): 把一个对象的操作代理到另一个对象
• 适配器模式(Adapter)：通过一个间接层适配统一接口
• 外观模式(Facade): 简化复杂对象的访问问题
• 享元模式(Flyweight): 通过对象复用(池)改善资源利用，比如连接池
• Model-View-Controller(MVC)：解耦展示逻辑和业务逻辑

代理模式：

• 把一个对象的操作代理到另个一对象
• 这里又要提到我们之前实现的Stack/Queue，把操作代理到 deque
• 通常使用 has-a 组合关系

from collections import deque

class Stack(object):  # 使用组合的例子

    def __init__(self):
        self._deque = deque()   # has a deque()

    def push(self, value):
        return self._deque.append(value)

    def pop(self):
        return self._deque.pop()

    def empty(self):
        return len(self._deque) == 0

适配器模式(Adapter)：

• 把不同对象的接口适配到同一个接口
• 想象一个多功能充电头，可以给不同的电器充电，充当了适配
• 当我们需要给不同的对象统一接口的时候可以使用适配器模式

# 适配器模式的例子
class Dog(object):
    def __init__(self):
        self.name = "Dog"

    def bark(self):
        return "woof!"


class Cat(object):
    def __init__(self):
        self.name = "Cat"

    def meow(self):
        return "meow!"


class Adapter:
    def __init__(self, obj, **adapted_methods):
        """We set the adapted methods in the object's dict"""
        self.obj = obj
        self.__dict__.update(adapted_methods)

    def __getattr__(self, attr):
        """All non-adapted calls are passed to the object"""
        return getattr(self.obj, attr)


objects = []
dog = Dog()
objects.append(Adapter(dog, make_noise=dog.bark))
cat = Cat()
objects.append(Adapter(cat, make_noise=cat.meow))
for obj in objects:
    print("A {0} goes {1}".format(obj.name, obj.make_noise()))

# 输出结果：
A Dog goes woof!
A Cat goes meow!

常见行为型设计模式

• 迭代器模式(Iterator): 通过统一的接口迭代对象
• 观察者模式(Observer):对象发生改变的时候，观察者执行相应动作
• 策略模式(Strategy): 针对不同规模输入使用不同的策略

迭代器模式：

• Python内置对迭代器模式的支持
• 比如我们可以用 for 遍历各种 Iterable 的数据类型
• Python里可以实现__next__和 __iter__ 实现迭代器

from collections import deque

class Stack(object):  # 使用组合的例子

    def __init__(self):
        self._deque = deque()   # has a deque()

    def push(self, value):
        return self._deque.append(value)

    def pop(self):
        return self._deque.pop()

    def empty(self):
        return len(self._deque) == 0

    def __iter__(self):
        res = []
        for i in self._deque:
            res.append(i)
        for i in reversed(res):
            yield i

s = Stack()
s.push(1)
s.push(2)
for i in s:
    print(i)

观察者模式：

• 发布订阅是一种最常用的实现方式
• 发布订阅用于解耦逻辑
• 可以通过回调等方式实现，当发生事件时，调用相应的回调函数

# 发布订阅模式

class Publisher:  # 发布者
    def __init__(self):
        self.observers = []  # 观察者

    def add(self, observer):  # 加入观察者
        if observer not in self.observers:
            self.observers.append(observer)
        else:
            print('Failed to add : {}').format(observer)

    def remove(self, observer):  # 移除观察者
        try:
            self.observers.remove(observer)
        except ValueError:
            print('Failed to remove : {}').format(observer)

    def notify(self):  # 调用观察者的回调
        [o.notify_by(self) for o in self.observers]


class Formatter(Publisher):  # 继承自发布者
    def __init__(self, name):
        super().__init__()
        self.name = name
        self._data = 0

    @property
    def data(self):
        return self._data

    @data.setter
    def data(self, new_value):
        self._data = int(new_value)
        self.notify()  # data 在被合法赋值以后会执行notify


class BinaryFormatter:
    """ 订阅者 """

    def notify_by(self, publisher):
        print("{}: '{}' has now bin data = {}".format(
            type(self).__name__,
            publisher.name,
            bin(publisher.data))
        )


def test():
    df = Formatter('formatter')  # 发布者
    bf = BinaryFormatter()  # 订阅者
    df.add(bf)  # 添加订阅者
    df.data = 3  # 设置的时候调用订阅者的notify_by

策略模式：

• 根据不同的输入采用不同的策略
• 比如买东西超过10个打八折，超过20个打七折
• 对外暴露统一的接口，内部采用不同的策略计算

# 策略模式

class Order:
    def __init__(self, price, discount_strategy=None):
        self.price = price
        self.discount_strategy = discount_strategy

    def price_after_discount(self):
        if self.discount_strategy:
            discount = self.discount_strategy(self)
        else:
            discount = 0
        return self.price - discount

    def __repr__(self):
        fmt = "<Price: {}, price after discount: {}>"
        return fmt.format(
            self.price, self.price_after_discount()
        )


def ten_percent_discount(order):
    return order.price * 0.10


def on_sale_discount(order):
    return order.price * 0.25 + 20


if __name__ == '__main__':
    order0 = Order(100)
    order1 = Order(100, discount_strategy=ten_percent_discount)
    order2 = Order(1000, discount_strategy=on_sale_discount)
    print(order0)
    print(order1)
    print(order2)

函数式编程

Python支持部分函数式编程特性

• 高阶函数: map/reduce/filter

闭包：

• 绑定了外部作用域的变量的函数
• 即使程序离开外部作用域，如果闭包仍然可见，绑定变量不会销毁
• 每次运行外部函数都会重新创建闭包

操作系统相关

Linux常见命令

如何查询 linux 命令的用法：

• 使用 man 命令查询用法。但是 man 手册比较晦涩
• 使用工具自带的help，比如 pip —help
• 这里介绍一个man 的替代工具 tldr。pip install tldr

掌握常见的文件操作工具

• chown/chmod/chgrp 改变所有者/改变模式/改变所有组
• ls/rm/cd/cp/mv/touch/rename/ln(软链接和硬链接) 等
• locate/find/grep 定位查找和搜索

文件或者日志查看工具：

• 编辑器 vi/nano/vim
• cat/head/tail 查看文件
• more/less 交互式查看文件

常见的进程操作工具:

• ps 查看进程
• kill 杀死进程
• top/htop 监控进程

常见的网络工具:

• ifconfig 查看网卡信息
• lsof/netstat 查看端口信息
• ssh/scp 远程登录/复制。tcpdump 抓包

操作系统线程和进程

进程和线程：

• 进程是对运行时程序的封装，是系统资源调度和分配的基本单位
• 线程是进程的子任务, cpu 调度和分配的基本单位，实现进程内并发
• 一个进程可以包含多个线程，线程依赖进程存在，并共享进程内存

线程安全：

• 一个操作可以在多线程环境中安全使用，获取正确的结果
• 线程安全的操作好比线程是顺序执行而不是并发执行的(i += 1)
• 一般如果涉及到写操作需要考虑如何让多个线程安全访问数据

了解线程同步的方式，如何保证线程安全：

• 互斥量（锁）: 通过互斥机制防止多个线程同时访问公共资源
• 信号量(Semphare):控制同一时刻多个线程访问同一个资源的线程数
• 事件(信号): 通过通知的方式保持多个线程同步

进程间通信的方式：

• 管道/匿名管道/有名管道(pipe)
• 信号(Signal): 比如用户使用Ctrl+c 产生 SIGINT 程序终止信号
• 消息队列 (Message)
• 共享内存(share memory)
• 信号量(Semaphore)
• 套接字 (socket)：最常用的方式，我们的 web 应用都是这种方式

python中如何使用多线程：

• threading.Thread 类用来创建线程
• start() 方法启动线程
• 可以用 join() 等待线程结束

import threading

lock = threading.Lock()

n = [0]


def foo():
    with lock:
        n[0] = n[0] + 1
        n[0] = n[0] + 1


threads = []
for i in range(5000):
    t = threading.Thread(target=foo)
    threads.append(t)

for t in threads:
    t.start()

print(n)

python中使用多进程：

multiprocessing模块

# 多进程
import multiprocessing


def fib(n):
    """worker function"""
    if n <= 1:
        return 1
    return fib(n - 1) + fib(n - 2)


if __name__ == '__main__':
    jobs = []
    for i in range(10, 20):
        p = multiprocessing.Process(target=fib, args=(i,))
        jobs.append(p)
        p.start()

操作系统内存管理机制

分页机制：

• 逻辑地址和物理地址分离的内存管理方案
• 程序的逻辑地址划分为固定大小的页(Page)
• 物理地址划分为同样大小的帧(Frame)
• 通过页表对应逻辑地址和物理地址

分段机制：

• 分段是为了满足代码的一些逻辑需求
• 数据共享，数据保护，动态链接等
• 通过段表实现逻辑地址和物理地址的映射关系
• 每个段内部是连续内存分配，段和段之间是离散分配的

分页和分段的区别：

• 页是出于内存利用率的角度提出的离散分配机制
• 段是出于用户角度，用于数据保护、数据隔离等用途的管理机制
• 页的大小是固定的，操作系统决定；段大小不确定，用户程序决定

什么是虚拟内存：

• 把一部分暂时不用的内存信息放到硬盘上
• 局部性原理，程序运行时候只有部分必要的信息装入内存
• 内存中暂时不需要的内容放到硬盘上
• 系统似乎提供了比实际内存大得多的容量，称之为虚拟内存

什么是内存抖动：

• 本质是频繁的页调度行为
• 频繁的页调度，进程不断产生缺页中断
• 置换一个页，又不断再次需要这个页
• 运行程序太多；页面替换策略不好。终止进程或者增加物理内存

python的垃圾回收机制

• 引用计数为主（缺点：循环引用无法解决）

  • 引用计数增加：当对象创建 a = 1；对象被应用 b = a；对象作为参数传递func(a)；对象存储在容器中 l = [a]
  • 引用计数减少：使用del a；引用指向了别的对象 b = None；离开对象的作用域(如函数执行结束)；从一个容器移除对象或者销毁容器

• 引入标记清除和分代回收解决引用计数的问题

  • 如 a=[1]，b=[2]，a.append(b)，b.append(a)，a和b的引用计数都是2，这时候无法将a和b的引用计数变为0，两个对象相互引用之后引用计数无法归零，出现了循环引用的问题。
  • 标记清除的算法，就是从根对象找到所有可达的点（上述例子中，就a和b相互引用，其他没有引用a和b的，是孤立的点，属于不可达的点），将所有不可达的点和可达的点标记，认为不可达的点没有其他地方引用了，就可以把它清除掉。

• 引用计数为主+标记清除和分代回收为辅

  • 分代回收，python把对象的生命周期分为3代（0，1，2），每一代都使用双向链表来标记对象。一开始 创建的对象称为第0代，python会针对第0、1、2代每隔一段时间会执行标记回收，这是有阈值的，会每隔多收时间清除第0代、清除第1代、清除第2代。
  • 比如说第0代执行完标记回收后，剩下的对象没有被回收，这个对象会转到更老的一代（第1代）。

  In [6]: import gc
  
  In [7]: gc.get_threshold()
  Out[7]: (700, 10, 10)
  
  In [8]: gc.get_threshold?
  Signature: gc.get_threshold()
  Docstring: Return the current collection thresholds.
  Type:      builtin_function_or_method

网络编程

网络协议TCP和UDP

浏览器输入一个url中间经历的过程

DNS缓存->DNS查询->拿到IP地址，浏览器就调用socket函数发起TCP请求（三次握手）->发起应用层协议，HTTP请求->Nginx反向代理->uwsgi/gunicorn->web app响应->TCP的四次挥手

TCP三次握手

TCP四次挥手：

TCP和UDP：

• TCP面向连接、可靠的、基于字节流
• UDP 无连接、不可靠、面向报文

HTTP协议

常见状态码：

• 1** 信息。服务器收到请求，需要请求者继续执行操作
• 2** 成功。操作被成功接受并处理
• 3** 重定向。需要进一步操作完成请求
• 4** 客户端错误。请求有语法错误或者无法完成请求
• 5** 服务器错误。服务器在处理请求的过程中发生错误

HTTP GET/POST 区别：

• Restful 语义上一个是获取，一个是创建
• GET 是幂等的，POST 非幂等
• GET请求参数放到url(明文),长度限制；POST 放在请求体，更安全

幂等性：

• 幂等方法是无论调用多少次都得到相同结果的 HTTP 方法
• 例如： a=4 是幂等的，但是 a += 4 就是非幂等的
• 幂等的方法客户端可以安全地重发请求

幂等方法：

什么是 HTTP 长连接：

• 连接：建立连接…数据传输…关闭连接(连接的建立和关闭开销大)
• 长连接：Connection: Keep-alive。保持 TCP 连接不断开
• 如何区分不同的 HTTP 请求呢？Content-Length | Transfer-Encoding: chunked

cookie和 session 区别：

• Session 一般是服务器生成之后给客户端（通过 url 参数或cookie)
• Cookie 是实现 session 的一种机制，通过 HTTP cookie 字段实现
• Session通过在服务器保存 sessionid 识别用户，cookie 存储在客户端

TCP socket编程

# 服务端
import socket
import time

s = socket.socket()
s.bind(('127.0.0.1', 8888))
s.listen()

while True:
    client, addr = s.accept()   # return conn, addr
    print(client)
    timestr = time.ctime(time.time()) + '\r\n'
    client.send(timestr.encode())   # send 参数 encode('utf8')
    client.close()
    

# 客户端
import socket

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('127.0.0.1', 8888))
s.sendall(b'Hello World')
data = s.recv(1024)
print(data.decode())
s.close()

如何使用 socket 发送HTTP请求:

• 使用 socket 接口发送 HTTP 请求
• HTTP建立在TCP基础之上
• HTTP是基于文本的协议

import socket

s = socket.socket()
s.connect(('www.baidu.com', 80))

http = b"GET / HTTP/1.1\r\nHost: www.baidu.com\r\n\r\n"
s.sendall(http)
buf = s.recv(1024)
print(buf)
s.close()

IO多路复用

一些常见的提升并发能力的方式：

• 多线程模型，创建新的线程处理请求
• 多进程模型，创建新的进程处理请求
• IO多路复用，实现单进程同时处理多个 socket 请求

什么是 IO 多路复用：

• 操作系统提供的同时监听多个 socket 的机制
• 为了实现高并发需要一种机制并发处理多个 socket
• Linux 常见的是 select/poll/epoll
• 可以使用单线程单进程处理多个 socket

阻塞IO服务端调用recvfrom的时候，经历了两个过程：

• 内核等待数据
• 数据从内核拷贝到用户进程

IO多路复用模型：

• 当执行select调用的时候，第一步还是内核等待数据，但是只会阻塞在这一步，一旦返回socket可读，就直接调用recvfrom拿到数据。
• select可以同时处理多个socket，有一个就绪应用程序代码就可以处理它

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

python如何实现IO多路复用：

• Python 的IO多路复用基于操作系统实现(select/poll/epoll)
• Python2 select 模块
• Python3 selectors 模块

selectors 模块 ：

• 事件类型：EVEN T_READ，EVENT_WRITE
• DefaultSelector类：自动根据平台选取合适的IO模型
  • register(fileobj, events, data=None)
  • unregister(fileobj)
  • modify(fileobj, envents, data=None)
  • select(timeout=None): returns[(key, events)]
  • close()

Python并发网络库：

• Tornado 并发网络库和同时也是一个web微框架
• Gevent 绿色线程(greenlet) 实现并发，猴子补丁修改内置 socket
• Asyncio Python3 内置的并发网络库，基于原生协程

Tornado 适用于微服务，实现 Restful 接口：

• 底层基于Linux 多路复用
• 可以通过协程或者回调实现异步编程

import tornado.ioloop
import tornado.web
from tornado.httpclient import AsyncHTTPClient


class APIHandler(tornado.web.RequestHandler):
    async def get(self):
        url = 'http://httpbin.org/get'
        http_client = AsyncHTTPClient()
        resp = await http_client.fetch(url)
        print(resp.body)
        return resp.body


def make_app():
    return tornado.web.Application([
        (r"/api", APIHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

gevnet：

• 基于轻量级绿色线程(greenlet)实现并发
• 需要注意 monkey patch，gevent 修改了内置的socket改为非阻塞
• 配合 gunicorn 和 gevent 部署作为 wsgi server

import gevent.monkey
gevent.monkey.patch_all()  # 修改内置的一些库非阻塞

import gevent
import requests


def fetch(i):
    url = 'http://httpbin.org/get'
    resp = requests.get(url)
    print(len(resp.text), i)


def asynchronous():
    threads = []
    for i in range(1, 10):
        threads.append(gevent.spawn(fetch, i))
    gevent.joinall(threads)


print('Asynchronous:')
asynchronous()

asyncio：

• 基于协程实现的内置并发网络库，协程+时间循环

# 基于 aiohttp 并发请求
import asyncio
from aiohttp import ClientSession   # pip install aiohttp

async def fetch(url, session):
    async with session.get(url) as response:
        return await response.read()

async def run(r=10):
    url = "http://httpbin.org/get"
    tasks = []

    async with ClientSession() as session:
        for i in range(r):
            task = asyncio.ensure_future(fetch(url, session))
            tasks.append(task)
        responses = await asyncio.gather(*tasks)
        for resp_body in responses:
            print(len(resp_body))


loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run())
loop.run_until_complete(future)

数据库相关

MySQL基础

事务：

• 事务是数据库并发控制的基本单位
• 事务可以看作是一系列SQL语句的集合
• 事务必须要么全部执行成功，要么全部执行失败（回滚）

Transaction示例：

# SQLAalchemy  事务回滚

session.begin()
try:
    item1 = session.query(Item).get(1)
    item2 = session.query(Item).get(2)
    item1.foo = 'bar'
    item2.bar = 'foo'
    session.commit()
except:
    session.rollback()
    raise

事务ACID特性：

• 原子性(Atomicity)：一个事务中所有操作全部完成或失败
• 一致性(Consistency): 事务开始和结束之后数据完整性没有被破坏
• 隔离性(Isolation): 允许多个事务同时对数据库修改和读写
• 持久性(Durability)：事务结束之后，修改是永久的不会丢失

事务的并发控制可能产生的问题：

• 幻读(phantom read):一个事务第二次查出现第一次没有的结果
• 非重复读(nonrepeatable read):一个事务重复读两次得到不同结果
• 脏读(dirty read):一个事务读取到另一个事务没有提交的修改
• 丢失修改(lost update): 并发写入造成其中一些修改丢失

四种事务隔离级别：为了解决并发控制异常，定义了4种事务隔离级别

• 读未提交(read uncommitted):别的事务可以读取到未提交改变
• 读已提交(read committed):只能读取已经提交的数据
• 可重复读(repeatable read):同一个事务先后查询结果一样 MySQL默认是这个
• 串行化(Serializable): 事务完全串行化的执行，隔离级别最高，执行效率最低

如何解决高并发场景下的插入重复：

高并发场景下，写入数据库会有数据重复问题

• 使用数据库的唯一索引
• 使用队列异步写入
• 使用 redis 等实现分布式锁

乐观锁和悲观锁：

• 悲观锁是先获取锁再进行操作。一锁二查三更新 select for update
• 乐观锁先修改,更新的时候发现数据已经变了就回滚(check and set)
• 使用需要根据响应速度、冲突频率、重试代价来判断使用哪一种

MySQL引擎

• MyISAM不支持事务 ，InnoDB支持事务
• MyISAM不支持外键，InnoDB支持外键
• MyISAM只支持表锁，InnoDB支持行锁和表锁

MySQL索引

什么是索引：

• 索引是数据表中一个或者多个列进行排序的数据结构
• 索引能够大幅提升检索速度
• 创建、更新索引本身也会耗费空间和时间

演示数据结构的网站

https://www.cs.usfca.edu/~galles/visualization/Algorithms.html

查找结构进化史：

• 线性查询：一个个找，实现简单；但是查找太慢
• 二分查找：需要的是有序的数组，实现比较简单；因为是有序的，插入会特别慢
• 哈希：查询快；但是占用空间(底层是利用冗余的数组来存储的)，不太适合存储大规模数据
• 二叉查找树：插入和查询很快（复杂度log(n)）；但是无法存储大规模数据， 会复杂度退化
• 平衡树：解决二叉查找树退化的问题，树是平衡的；缺点就是节点非常多的时候，树高会很高
• 多路查找树：解决平衡树 树高的问题，一个父亲多个孩子节点；节点过多树高不会特别深
• 多路平衡查找树：B-Tree 每个节点最多 m(m>=2)个孩子,称为m 阶或者度
  • 叶节点具有相同的深度
  • 节点中的数据 key 从左到右是递增的

b树 进行范围查找比较困难

B+Tree：

• B+树是B-Tree的变形
• Mysql 实际使用的 B+Tree作为索引的数据结构
• 只在叶子节点带有指向记录的指针 (为什么？可以增加树的度)
  • B树的每个结点都存储了key和data，B+树的data存储在叶子节点上。
    节点不存储data，这样一个节点就可以存储更多的key。可以使得树更矮，所以IO操作次数更少
  • 不是阶越大就越好。硬盘是以块来存储的，阶是由磁盘块的大小来确定的。为了让操作系统更好的读取和缓存数据，以磁盘块的大小来确定B+树的阶
• 叶子结点通过指针相连。为什么？实现范围查询

MySQL索引类型：

• 普通索引 (CREATE INDEX)
• 唯一索引，索引列的值必须唯一 (CREATE UNIQUE INDEX)
• 多列索引
• 主键索引 (PRIMARY KEY)，一个表只能有一个
• 全文索引（FULLTEXT INDEX)，InnoDB 不支持

什么时候创建索引：

• 经常用作查询条件的字段(WHERE条件)
• 经常用作表连接的字段
• 经常出现在 order by, group by 之后的字段

MySQL索引最佳实践：

• 非空字段 NOT NULL，Mysql 很难对空值作查询优化
• 区分度高，离散度大（如枚举字段，就几个值，没必要创建索引），作为索引的字段值尽量不要有大量相同值
• 索引的长度不要太长(比较耗费时间)

索引什么时候失效：

• 以 % 开头的 LIKE 语句，模糊搜索
• 出现隐式类型转换，类型不匹配
• 没有满足最左前缀原则

聚集索引和非聚集索引：

• 聚集还是非聚集指的是B+Tree 叶节点存的是指针还是数据记录
• MyISAM索引和数据分离，使用的是非聚集索引
• InnoDB数据文件就是索引文件，主键索引就是聚集索引

非聚集索引：

聚集索引：

排查慢查询：

• 慢查询通常是缺少索引，索引不合理或者业务代码实现导致
• slow_query_log_file 开启并且查询慢查询日志
• 通过 explain 排查索引问题
• 调整数据修改索引；业务代码层限制不合理访问

缓存机制和Redis

什么是缓存？为什么要使用缓存？

• 缓解关系数据库(常见的是Mysql)并发访问的压力：热点数据
• 减少响应时间：内存 IO 速度比磁盘快
• 提升吞吐量：Redis 等内存数据库单机就可以支撑很大并发

Redis和Memcached的区别：

Redis常用数据类型和使用场景：

• String(字符串):用来实现简单的 KV 键值对存储，比如计数器
• List(链表):实现双向链表，比如用户的关注，粉丝列表
• Hash(哈希表): 用来存储彼此相关信息的键值对
• Set(集合): 存储不重复元素，比如用户的关注者
• Sorted Set(有序集合): 实时信息排行榜

Redis的底层实现：

• String: 整数或者sds(Simple Dynamic String)
• List: ziplist或者double linked list
  • ziplist(压缩列表) 通过一个连续的内存块实现list结构，其中的每个entry节点头部保存前后节点长度信息，实现双向链表功能。
• Hash: ziplist 或者 hashtable（哈希表）
• Set: intset 或者 hashtable
• SortedSet: skiplist 跳跃表

跳跃表：

sorted set为了简化实现，使用skiplist而不是平衡树实现

Redis持久化方式：

• RDB 快照方式：把数据快照放在磁盘二进制文件中，dump.rdb，在指定时间间隔把Redis数据库状态保存到一个压缩的二级制文件中
  • 相当于记录一个一个时间段的快照，可以很好的恢复到某个时间段的版本
  • 在某个时间段内，数据库宕机了，很有可能会丢失整个时间间隔之内的数据。
• AOF(Append Only File)：每一个写命令追加到 appendonly.aof中
  • 缺点就是每个文件会比较大

https://www.cnblogs.com/madashu/p/12832766.html

Redis事务：

• 将多个请求打包，一次性、按序执行多个命令的机制
• Redis 通过 MULTI, EXEC, WATCH 等命令实现事务功能
• Python redis.py pipeline=conn.pipeline(transaction=True)
• redis原子性不支持回滚；一致性需要使用watch来监听key，如果修改的话就是执行失败；隔离性，redis单线程的，本身支持；持久性，RDB和AOF并不能严格的保证持久化。

Redis如何实现分布式锁：

• 使用setnx实现加锁，可以同时通过expire添加超时时间
• 锁的 value 值可以使用一个随机的 uuid 或者特定的命名
• 释放锁的时候，通过uuid 判断是否是该锁，是则执行delete释放锁

常用的缓存使用模式：

• Cache Aside: 同时更新缓存和数据库
• Read/Write Through: 先更新缓存，缓存负责同步更新数据库
• Write Behind Caching: 先更新缓存，缓存定期异步更新数据库

数据库与缓存之间的数据一致性：

• 先更新数据库后更新缓存，并发操作可能导致缓存读取的是脏数据
• 一般先更新数据库然后删除缓存，下次读取的时候再重建缓存

缓存穿透：

• 大量查询不到的数据的请求落到后端数据库，数据库压力增大，如爬虫按照自增id爬取数据，很多id数据库都没有。
• 由于大量缓存查不到就去数据库取，数据库也没有要查的数据
• 解决：
  • 对于没查到返回为 None 的数据也缓存
  • 插入数据的时候删除相应缓存，或者设置较短的超时时间

缓存击穿：

• 某些非常热点的数据 key过期，大量请求打到后端数据库
• 热点数据 key 失效导致大量请求打到数据库增加数据库压力
• 解决：
  • 分布式锁：获取锁的线程从数据库拉数据更新缓存，其他线程等待
  • 异步后台更新：后台任务针对过期的 key 自动刷新

缓存雪崩：

• 缓存不可用或者大量缓存key同时失效，大量请求直接打到数据库
• 解决：
  • 多级缓存：不同级别的 key 设置不同的超时时间
  • 随机超时：key 的超时时间随机设置，防止同时超时
  • 架构层：提升系统可用性。监控、报警完善

python web框架

WSGI：

• Python Web Server Gateway Interface (pep3333)，在web server和web应用之间提供一个标准的接口，从而让web应用部署在任意的web server上
• 解决 Python Web Server 乱象 mod_python, CGI, FastCGI 等
• 描述了Web Server(Gunicorn/uWSGI)如何与 web 框架(Flask/Django)交互，Web 框架如何处理请求

# WSGI定义
def application(environ, start_response)

# application就是WSGI app，一个可调用对象
# environ参数：一个包含WSGI环境信息的字典，由WSGI服务器提供，常见的key有PATH_INFO，QUERY_STRING等
# start_response: 生成WSGI响应的回调函数，接收两个参数，status和headers
# 函数返回响应体的可迭代对象

编写一个兼容WSGI的web应用：

def myapp(environ, start_response):
    print(environ['QUERY_STRING'])
    status = '200 OK'
    headers = [('Content-Type', 'text/html; charset=utf8')]

    start_response(status, headers)
    return [b'<h1>Hello World</h1>']  # 可迭代对象，返回 字节


if __name__ == '__main__':
    from wsgiref.simple_server import make_server
    httpd = make_server('127.0.0.1', 8888, myapp)
    httpd.serve_forever()

封装成类：

class Application(object):

    def __init__(self, routers, **kwargs):
        self.routers = routers

    def __call__(self, environ, start_response):
        try:
            request = Request(environ)
            callback, args = routers.match(request.path)
            response = callback(request, *args)
        except NotFoundError:
            response = Response("<h1>Not found</h1>", status=404)
        start_response(response.status, response.headers.items())
        return iter(response)

常见的python web 框架

• Django: 大而全，内置 ORM、Admin 等组件，第三方插件较多
• Flask:微框架，插件机制，比较灵活
• Tornado:异步支持的微框架和异步网络库

MVC：

• Model: 负责业务对象和数据库的交互(ORM)
• View：负责与用户的交互展示
• Controller：接收请求参数调用模型和视图完成请求

Web 框架一般有哪些组件：

• 中间件，用于请求之前和请求之后做一些处理（比如记录日志等）
• 路由, 表单验证, 权限认证, ORM, 视图函数, 模板渲染, 序列化
• 第三方插件：Redis连接，RESTful支持等

Gunicorn: Python WSGI HTTP Server

• 纯 Python 编写的高性能的 WSGI Server
• pre-fork 预先分配多个 worker 进程处理请求(master-slave)
• 多种 worker 支持：Sync/Async(Gevent)/Tornado/AsyncIO
• gunicorn -w 4 myapp:app

常见的web 安全问题

• SQL注入
• XSS(跨站脚本攻击, Cross-Site Scripting)
• CSRF(跨站请求伪造, Cross-site request forgery)

SQL注入

• 通过构造特殊的输入参数传入Web应用，导致后端执行了恶意 SQL
• u通常由于程序员未对输入进行过滤，直接动态拼接 SQL 产生
• 可以使用开源工具 sqlmap, SQLninja 检测

防范 SQL 注入：

• 对输入参数做好检查（类型和范围）；过滤和转义特殊字符
• 不要直接拼接 sql，使用 ORM 可以大大降低 sql 注入风险
• 数据库层：做好权限管理配置；不要明文存储敏感信息

XSS跨站脚本攻击

• 恶意用户将代码植入到提供给其他用户使用的页面中，未经转义的恶意代码输出到其他用户的浏览器被执行
• 用户浏览页面的时候嵌入页面中的脚本(js)会被执行，攻击用户
• 主要分为两类：反射型(非持久型)，存储型(持久型)

XSS的危害：

• 盗用用户 cookie，获取敏感信息
• 利用用户私人账号执行一些违法操作，比如盗取个人或者商业资料，执行一些隐私操作
• 甚至可以在一些访问量很大的网站上实现DDoS 攻击

如何防范XSS：

• 过滤(输入和参数)。对敏感标签 <script> <img> <a>等进行过滤
• 转义。对常见符号(“&”, “<” and “>)转义（python3 html.escape)
• 设置HttpOnly 禁止浏览器访问和操作 Document.cookie

CSRF跨站请求伪造

• 利用网站对已认证用户的权限去执行未授权的命令的一种恶意攻击
• 攻击者会盗用你的登录信息，以你的身份模拟发送请求
• web 身份认证机制只能识别一个请求是否来自某个用户的浏览器，但是无法保证请求是用户自己或者批准发送的

CSRF产生的条件：

• 受害者已经登录到了目标网站并且没有退出(保持登录状态)
• 受害者访问了攻击者发布的链接或者表单
• 二者必须缺一不可

如何防范CSRF：不要在 GET 请求里有任何数据修改操作

• 令牌同步(Synchronizer token pattern，简称STP)：在用户请求的表单中嵌入一个隐藏的csrf_token，服务端验证其是否与 cookie 中的一致（基于同源策略其他网站是无法获取cookie中的csrf_token)
• 如果是 js 提交需要先从cookie获取csrf_token作为 X-CSRFToken请求头提交提交
• 其他：检测来源HTTP Referer(容易被伪造)；验证码方式(安全但是繁琐)

前后端分离与 RESTful

什么是前后端分离：

• 后端只负责提供数据接口，不再渲染模板，前端获取数据并呈现
• 前后端解耦，接口复用（前端和客户端公用接口），减少开发量
• 各司其职，前后端同步开发，提升工作效率。定义好接口规范
• 更有利于调试(mock)、测试和运维部署

什么是RESTful：Representational State Transfer

• 表现层状态转移，由 HTTP 协议的主要设计者Roy Fielding提出
• 资源(Resources)，表现层(Representation)，状态转化(State Transfer)
  • Resources(资源): 使用 URI 指向的一个实体
  • Representation(表现层): 资源的表现形式，比如图片、HTML 文本等
  • State Transfer(状态转化): GET、POST、PUT、DELETE HTTP 动词来操作资源，实现资源状态的改变
• 是一种以资源为中心的 web软件架构风格，可以用 Ajax 和 RESTful web服务构建应用

RESTful的准则：

• 所有事物抽象为资源(resource)，资源对应唯一的标识(identifier)
• 资源通过接口进行操作实现状态转移，操作本身是无状态的
• 对资源的操作不会改变资源的标识

RESTful 风格的 API 接口：

• 通过 HTTP GET/POST/PUT/DELETE 获取/新建/更新/删除 资源
• 一般使用 JSON 格式返回数据

如何设计RESTful风格的API：

HTTP方法	URL	动作
GET	http://[hostname]/api/users	检索用户列表
GET	http://[hostname]/api/users/[user_id]	检索单个用户
POST	http://[hostname]/api/users	创建新用户
PUT	http://[hostname]/api/users/[user_id]	更新用户信息
DELETE	http://[hostname]/api/users/[user_id]	删除用户

使用tornado实现RESTful API：

user_handler.py:

import tornado.web
from tornado.escape import json_encode
from models.user import UserModel


class UserListHandler(tornado.web.RequestHandler):
    def get(self):   # 批量获取用户
        users = UserModel.get_all()
        self.write(json_encode(users))

    def post(self):
        name = self.get_argument('name')
        age = self.get_argument('age')
        UserModel.create(name, age)
        resp = {'status': True, 'msg': 'create success'}
        self.write(json_encode(resp))


class UserHandler(tornado.web.RequestHandler):

    def get(self, user_id):
        try:
            user = UserModel.get(int(user_id))
        except KeyError:
            return self.set_status(404)
        self.write(json_encode(user))

    def put(self, user_id):
        age = self.get_argument('age')
        UserModel.update(int(user_id), age)
        resp = {'status': True, 'msg': 'update success'}
        self.write(json_encode(resp))

    def delete(self, user_id):
        UserModel.delete(int(user_id))
        resp = {'status': True, 'msg': 'delete success'}
        self.write(json_encode(resp))

user_model.py:

class UserModel(object):

    users = {
        1: {'name': 'zhang', 'age': 10},
        2: {'name': 'wang', 'age': 12},
        3: {'name': 'li', 'age': 20},
        4: {'name': 'zhao', 'age': 30},
    }

    @classmethod
    def get(cls, user_id):
        return cls.users[user_id]

    @classmethod
    def get_all(cls):
        return list(cls.users.values())

    @classmethod
    def create(cls, name, age):
        user_dict = {'name': name, 'age': age}
        max_id = max(cls.users.keys()) + 1
        cls.users[max_id] = user_dict

    @classmethod
    def update(cls, user_id, age):
        cls.users[user_id]['age'] = age

    @classmethod
    def delete(cls, user_id):
        if user_id in cls.users:
            return cls.users.pop(user_id)

系统设计

什么是系统设计：

• 是一个定义系统架构、模块、接口和数据满足特定需求的过程
• 比如一个短网址服务、评论服务、Feed流系统、抢红包系统
• 微服务架构很多系统被按照业务拆分，需要单独设计一个系统服务

系统设计的要素：

• 使用场景和限制条件
  • 这个系统是在什么地方使用的？比如短网址系统提供给站内各种服务生成短网址
  • 限制条件：用户多少？至少要能支撑多少服务
  • 估算并发qps：峰值qps是多少？平均qps是多少？
• 数据存储设计，数据库选型
  • 按需设计数据表，需要哪些字段，使用什么类型？数据的增长规模是怎样的
  • 数据库选型：是否需要持久化？使用关系型还是NoSQL？
  • 如何优化？如何设计索引？是否可以使用缓存？
• 算法模块设计，算法是解决问题的核心。程序=算法+数据结构。系统=服务+存储
  • 需要哪些接口？接口如何设计
  • 使用什么算法或者模型？
  • 不同实现方式之间的优劣对比，如何取舍？

短网址系统的设计和实现

如何设计和实现一个短网址系统？

• 什么是短网址系统？包含哪些功能？
• 短网址系统的存储设计？需要存储哪些字段
• 如何设计算法生成短网址？

什么是短网址系统：

• 把长网址转换成短网址的服务
• 比如 https://bitly.com/
• 转换之后的网址后缀不超过7位

场景和限制：

• 场景：提供短网址服务为公司其他个业务服务
• 功能：一个长网址转成短网址并存储；根据短网址还原长网址
• 要求短网址的后缀不超过7位
• 预估峰值插入请求数量级：数百；查询请求数量级：数千

数据存储设计：

• 使用MySQL

• 字段

  • id
  • token 存储短网址生成的token，给token加入索引
  • url（原来的网址）
  • created_time

算法实现设计

• 两个API：long2short_url，short2long_url

短网址生成算法：

• md5摘要算法，取出前7个字符。但是会有冲突
• 26个小写字母+26个大写字母+10个数字 ，26+26+10=62
  • 类似于62进制的数字
  • 十进制->62进制，可以类比10进制->2进制
  • 递增序列算法
• 自增id 需要一个计数器 使用Redis的 incr

# python实现10进制到2进制的转化

def mybin(num):  # 10进制->2进制串
    if num == 0:
        return 0
    res = []
    while num:
        num, rem = divmod(num, 2)
        res.append(str(rem))
    return ''.join(reversed(res))

CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

def encode(num):
    if num == 0:
        return CHARS[0]
    res = []
    while num:
        num, rem = divmod(num, len(CHARS))  # 62
        res.append(CHARS[rem])
    return ''.join(reversed(res))


print(encode(1))
print(encode(61))

实现：

创建的数据库表：

CREATE TABLE short_url (
  id bigint unsigned NOT NULL AUTO_INCREMENT,
  token varchar(10),
  url varchar(2048),
  created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_token` (`token`)
);

short_url_app.py

"""
短网址生成代码。Flask 演示
依赖本地 mysql 和 redis
"""
import os

from flask import Flask, jsonify, render_template, request
from flask_mysqldb import MySQL
from flask.ext.redis import FlaskRedis

app = Flask(__name__)
app.config['MYSQL_USER'] = 'root'
app.config['MYSQL_PASSWORD'] = os.getenv('MYSQL_PASS') # export your password
app.config['MYSQL_DB'] = 'test'
app.config['MYSQL_CURSORCLASS'] = 'DictCursor'

mysql = MySQL(app)
redis_store = FlaskRedis(app)


CHARS = "abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ0123456789"

def encode(num):
    if num == 0:
        return CHARS[0]
    res = []
    while num:
        num, rem = divmod(num, len(CHARS))  # 62
        res.append(CHARS[rem])
    return ''.join(reversed(res))

@app.route('/shorten', methods=['POST'])
def shorten_url():
    long_url = request.json['url']
    index = int(redis_store.incr('SHORT_CNT'))
    token = encode(index)
    sql = "INSERT INTO short_url(token, url) VALUES(%s, %s)"
    cur = mysql.connection.cursor()
    cur.execute(sql, (token, long_url))
    mysql.connection.commit()
    short_url = 'https://short.com/' + token
    return jsonify(url=short_url)


@app.route('/')
def index():
    return render_template('index.html')


if __name__ == '__main__':
    app.run(debug=1)

templates/index.html

<!DOCTYPE html>
<html>
<head>
    <title>短网址服务</title>
    <meta name="viewport" content="width=device-width, initial-scale=1">

    <script src="https://cdn.bootcss.com/jquery/3.1.0/jquery.min.js"></script>
    <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/materialize/0.98.0/css/materialize.min.css">
    <link rel=stylesheet type=text/css href="{{ url_for('static', filename='css/custom.css') }}">
</head>

<script type="text/javascript">
 $(function() {
     $('#submitButton').click(function() {
         $.ajax({
             type: "POST",
             url: "/shorten",
             data: JSON.stringify({'url' : $('#url').val()}),
             success: returnSuccess,
             dataType: 'json',
             contentType: "application/json",
         });
     });
});

function returnSuccess(data, textStatus, jqXHR) {
    if(data.url) {
        $('#url-result').text(data.url);
        $('#url').val("");
        $("a").attr("href", data.url)
    } else {
        $('#url-result').text("请输入一个URL!");
    }
}
</script>

<body>
    <div id="wrap">
        <header>
           <h1><a href="">短网址服务</a></h1>
           <p>请输入一个网址</p>
        </header>
        <div class="container">
            <div class="row">
                <div class="col s12">
                    <input type="text" name="url" id="url" class="form-control input-sm" placeholder="请输入一个合法的网址" />
                </div>
            </div>
            <div class="row">
                <div class="col s4 offset-s4">
                    <button id="submitButton" class="waves-effect waves-light btn-large blue darken-1">生成</button>
                </div>
            </div>
            <div class="row">
                <div class="col s12">
                    <div class="panel-footer">
                        <h4>
                            <a href="#" id="url-result">输入 URL</a>
                        </h4>
                    </div>
                </div>
            </div>
        </div>
    </div>
</body>
</html>


<!-- 短网址服务，前端代码参考 aiohttp demo -->

static/css/custom.css

* {
    margin: 0;
}

html, body {
    height: 100%;
}

#content {
    margin-top: 20px;
    margin-bottom: 60px;
}

header {
    margin-bottom: 30px;
    padding-bottom: 10px;
    clear: both;
}

#wrap {
    margin: 5px 10px -50px 10px;
    padding: 10px;
    text-align: center;
    min-height: 100%;
}
#url-result {
    color: #039be5;
    display: hidden;
}

.btn {
    float: right;
}

footer {
    color: #039be5;
    text-align: center;
    height: 50px;
    line-height: 50px;
}

footer a{
    color: #039be5;
}