谈谈监控的选型相关问题

传统的主机监控类

zabbix

很早有许多使用 zabbix 做相关监控的

核心组件的话 是agent server 模式的
agent负责采集数据 主动或者被动的方式发送到 server/proxy
为了扩展 还可以进行自定义脚本的监控等等

数据存储默认是选择的mysql 提供一个php web的页面来做查询

因为是使用关系型数据库 因此在大规模集群的时候 有点捉襟见肘

4.2版本之后开始采用时序的数据库 但是成熟度不算很高

open-falcon

是小米成熟的开源项目 目前在市面上也有很多公司选用

小米 滴滴（在此基础上做了夜莺系统拥抱开源） 美团等

采用golang 开发的daemon 程序 运行在每台linux机器上 采集主机的各种指标

heartbeat server 心跳服务 周期性的通过RPC 的方式上报信息 主机名 ip agent版本 插件版本等

transfer 负责接收发送的监控数据 并对数据进行数据整理 过滤之后通过一致性哈希发送到judge/graph

graph 是基于RRD的数据上报 归档 存储组件

judge 模块 触发用户的告警规则 满足的情况下 会触发邮件 微信或者是回调接口 为了避免重复告警引入redis暂存告警 从而完成告警的合并和抑制

dashboard 面向用户的监控数据查询和告警配置页面

腾讯蓝鲸

腾讯开源的一整套系统 相对来说 是比较重的资产 要全套使用 系统配置相当之繁琐

prometheus

2016 年 CNCF 的第二大开源项目

通过http 周期性抓取被监控组件的状态 任意组件只要提供对应的http接口并且符合定义的数据格式 就可以接入到监控之中

prometheus server pull 多个 exporter 的数据 pull这种方式降低了客户端的复杂性 不需了解服务端情况

监控数据如果达到告警阀值 会将告警发送到 altermanger 邮件或者webhook的方式来通知消息

页面展示的话 一般选择还是grafana

一般情况下拉取的数据还是会存储在本地的 tsdb

开发语言上来看 java占据业务开发 c占据底层开发 golang 在中间件的开发需求 在开源中间件重应用广泛

再就是容器角度 在云原生蓬勃发展的时代 成为主导容器监控的标配

python项目的高并发选择

一般的高并发选择 gunicorn/uwsgi + nginx

改框架 搭配其他手段来提升并发 最大的开销还是在io 上

cdn 缓存 负载均衡 缓存代理等手段 辅助提高

百万并发问题不大

拆开来看的话 就是 三种情况

• 侧重读
• 侧重写
• 读写并重

读的情况下

搜索引擎 商品搜索

策略类
加缓存 空间换时间 本地或者远程缓存/主从/cdn动静分离
并发读 异步rpc 冗余请求

侧重写的情况

广告计算

策略

数据分片
任务分片
任务和数据分片相结合
异步化 异步处理/异步罗盘 凡是不阻碍主流程的业务逻辑都可以进行异步化
批量处理
串行化+多进程单线程+异步IO nginx/redis

读写并重 电商的缓存和秒杀系统 支付系统和红包 IM 微博 朋友圈

协程库

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

import select

from collections import deque
from socket import socket, AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR

def create_listen_socket(bind_addr='0.0.0.0', bind_port=55555, backlogs=102400):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind((bind_addr, bind_port))
    sock.listen(backlogs)
    return sock

class Future:

    def __init__(self, loop):
        self.loop = loop
        self.done = False
        self.result = None
        self.co = None

    def set_coroutine(self, co):
        self.co = co

    def set_result(self, result):
        self.done = True
        self.result = result

        if self.co:
            self.loop.add_coroutine(self.co)

    def __await__(self):
        if not self.done:
            yield self
        return self.result

class AsyncSocket:

    def __init__(self, sock, loop):
        sock.setblocking(False)

        self.sock = sock
        self.loop = loop

    def fileno(self):
        return self.sock.fileno()

    def create_future_for_events(self, events):
        future = self.loop.create_future()

        def handler(fileno, active_events):
            loop.unregister_from_polling(self.fileno())
            future.set_result(active_events)

        self.loop.register_for_polling(self.fileno(), events, handler)

        return future

    async def accept(self):
        while True:
            try:
                sock, addr = self.sock.accept()
                return AsyncSocket(sock=sock, loop=self.loop), addr
            except BlockingIOError:
                future = self.create_future_for_events(select.EPOLLIN)
                await future

    async def recv(self, bufsize):
        while True:
            try:
                return self.sock.recv(bufsize)
            except BlockingIOError:
                future = self.create_future_for_events(select.EPOLLIN)
                await future

    async def send(self, data):
        while True:
            try:
                return self.sock.send(data)
            except BlockingIOError:
                future = self.create_future_for_events(select.EPOLLOUT)
                await future

class EventLoop:

    def __init__(self):
        self.epoll = select.epoll()

        self.runnables = deque()
        self.handlers = {}

    def create_future(self):
        return Future(loop=self)

    def create_listen_socket(self, bind_addr, bind_port, backlogs=102400):
        sock = create_listen_socket(bind_addr, bind_port, backlogs)
        return AsyncSocket(sock=sock, loop=loop)

    def register_for_polling(self, fileno, events, handler):
        print('register fileno={} for events {}'.format(fileno, events))
        self.handlers[fileno] = handler
        self.epoll.register(fileno, events)

    def unregister_from_polling(self, fileno):
        print('unregister fileno={}'.format(fileno))
        self.epoll.unregister(fileno)
        self.handlers.pop(fileno)

    def add_coroutine(self, co):
        self.runnables.append(co)

    def run_coroutine(self, co):
        try:
            future = co.send(None)
            future.set_coroutine(co)
        except StopIteration as e:
            print('coroutine {} stopped'.format(co.__name__))

    def schedule_runnable_coroutines(self):
        while self.runnables:
            self.run_coroutine(co=self.runnables.popleft())

    def run_forever(self):
        while True:
            self.schedule_runnable_coroutines()

            events = self.epoll.poll(1)
            for fileno, event in events:
                handler = self.handlers.get(fileno)
                if handler:
                    handler(fileno, events)

class TcpServer:

    def __init__(self, loop, bind_addr='0.0.0.0', bind_port=55555):
        self.loop = loop
        self.listen_sock = self.loop.create_listen_socket(bind_addr=bind_addr, bind_port=bind_port)
        self.loop.add_coroutine(self.serve_forever())

    async def serve_client(self, sock):
        while True:
            data = await sock.recv(1024)
            if not data:
                print('client disconnected')
                break

            await sock.send(data.upper())

    async def serve_forever(self):
        while True:
            sock, (addr, port) = await self.listen_sock.accept()
            print('client connected addr={} port={}'.format(addr, port))

            self.loop.add_coroutine(self.serve_client(sock))

if __name__ == '__main__':
    loop = EventLoop()
    server = TcpServer(loop=loop)
    loop.run_forever()

我们模仿常见协程库，引入 Future ，代表一个在未来才能获取到的数据。Future 一般由协程创建，典型的场景是这样的：协程在等待一个 IO 事件，这时它便创建一个 Future 对象，并把执行权归还给事件循环。

例子中的 Future 类，有 4 个重要的属性：

• loop ，当前事件循环对象；
• done ，标识目标数据是否就绪；
• result ，目标数据；
• co ，关联协程，Future 就绪后，事件循环 loop 将把它放入可执行队列重新调度；

注意到，Future 是一个 可等待对象 ( awaitable )，它实现了 await 方法。当数据未就绪时，通过 yield 让出执行权，这时事件循环将协程记录在 Future 中。当数据就绪后，事件循环将协程放回可执行队列重新调度。

协程库还将套接字进行 异步化 封装，抽象出 AsyncSocket 类，接口与原生 socket 对象类似。除了保存原生 socket 对象，它还保存事件循环对象，以便通过事件循环订阅 IO 事件。

create_future_for_events 方法创建一个 Future 对象，来等待一个不知何时发生的 IO 事件。创建完 Future 对象后，进一步调用 loop 相关方法，将感兴趣的 IO 事件注册到 epoll 。当相关事件就绪时，事件循环将执行回调函数 handler ，它解除 epoll 注册，并将活跃事件作为目标数据设置到 Future 上(注意 set_result 将唤醒协程)。

然后是套接字系列操作函数，以 accept 为例，它不断尝试调用原生套接字，而原生套接字已被设为非阻塞。如果套接字已就绪，accept 将直接返回新连接，协程无须等待。

否则，accept 方法抛出 BlockingIOError 异常。这时，协程调用 create_future_for_events 方法创建一个 Future 订阅读事件( EPOLLIN )，并等待事件到达。

recv 、send 方法封装也是类似的，不同的是 send 需要订阅 可写事件 ( EPOLLOUT )。

好了，终于来到协程库了主角事件循环 EventLoop 对象了，它有 3 个重要属性：

• epoll ，这是一个 epoll 描述符，用于订阅 IO 事件；
• runnables ，可执行协程队列；
• handlers ，IO 事件回调处理函数映射表；

register_for_polling 方法注册感兴趣的 IO 事件和处理函数，它以文件描述符为键，将处理函数记录到映射表中，然后调用 epoll 完成事件订阅。unregister_from_polling 方法则刚好相反，用于取消注册。

add_coroutine 将一个可运行的协程加入队列。run_coroutine 则调度一个可执行协程，它调用 send 将执行权交给协程。如果协程执行完毕，它将输出提示；协程需要等待时，会通过 yield 归还执行权并提交 Future 对象，它将协程记录到 Future 上下文。schedule_runnable_coroutines 将可执行协程逐个取出并调度，直到队列为空。

run_forever 是事件循环的主体逻辑，这是一个永久循环。每次循环时，先调度可执行协程；然后通过 poll 等待协程注册的 IO 事件；当有新事件到达时，取出回调函数 handler 函数并调用。

TcpServer 只是一个普通的协程式应用，无须赘述。接下来，我们逐步分析，看看程序启动后都发生什么事情：

1. 创建事件循环 EventLoop 对象，它将创建 epoll 描述符；
2. 创建 TcpServer 对象，它通过事件循环 loop 创建监听套接字，并将 serve_forever 协程放入可执行队列；
3. 事件循环 loop.run_forever 开始执行，它先调度可执行队列；
4. 可执行队列一开始只有一个协程 TcpServer.serve_forever ，它将开始执行(由 run_coroutine 驱动)；
5. 执行权来到 TcpServer.serve_forever 协程，它调用 AsyncSocket.accept 准备接受一个新连接；
6. 假设原生套接字未就绪，它将抛出 BlockingIOError 异常；
7. 由于 IO 未就绪，协程创建一个 Future 对象，用来等待一个未来的 IO 事件( AsyncSocket.accept )；
8. 于此同时，协程调用事件循环 register_for_polling 方法订阅 IO 事件，并注册回调处理函数 handler ；
9. future 是可以个可等待对象，await future 将执行权交给它的 __await__ 函数；
10. 由于一开始 future 是未就绪的，这时 yield 将协程执行逐层归还给事件循环，future 对象也被同时上报；
11. 执行权回到事件循环，run_coroutine 收到协程上报的 future 后将协程设置进去，以便 future 就绪后重新调度协程；
12. 可执行队列变空后，事件循环开始调用 epoll.poll 等待协程注册的 IO 事件( serve_forever )；
13. 当注册事件到达后，事件循环取出回调处理函数并调用；
14. handler 先将套接字从 epoll 解除注册，然后调用 set_result 将活跃事件作为目标数据记录到 future 中；
15. set_result 将协程重新放回可执行队列；
16. IO 事件处理完毕，进入下一次事件循环；
17. 事件循环再次调度可执行队列，这时 TcpServer.serve_forever 协程再次拿到执行权；
18. TcpServer.serve_forever 协程从 yield 语句恢复执行，开始返回目标数据，也就是先前设置的活跃事件；
19. AsyncSocket.accept 内 await future 语句取得活跃事件，然后循环继续；
20. 循环再次调用原生套接字，这时它早已就绪，得到一个新套接字，简单包装后作为结果返回给调用者；
21. TcpServer.serve_forever 拿到代表新连接的套接字后，创建一个 serve_client 协程并交给事件循环 loop ；
22. TcpServer.serve_forever 进入下一次循环，调用 accept 准备接受下一个客户端连接；
23. 如果监听套接字未就绪，执行权再次回到事件循环；
24. 事件循环接着调度可执行队列里面的协程，TcpServer.serve_client 协程也开始执行了；
25. etc

这看着就像一个精密的机械装置，有条不紊的运行着，环环相扣！

重新看看 GIL的问题

list 、 dict 等内建对象是 线程安全 的吗？

在 Python 层面，list 、dict 等内建对象是线程安全的，这是最基本的常识。研究 list、dict 等内建对象源码时，我们并没有看到任何 互斥锁 的痕迹，这多少有点令人意外。

Python 虚拟机维护了一个 全局锁 ，这就是众所周知的 GIL。Python 线程想在虚拟机中执行字节码，必须取得全局锁。这样一来，不管任何时刻，只有一个线程在虚拟机中运行。那么，虚拟机如何交替执行不同线程呢？

Python 线程调度实现方式参考了操作系统进程调度中 时间片 的思路，只不过将时间片换成 字节码 。当一个线程取得 GIL 全局锁并开始执行字节码时，对已执行字节码进行计数。当执行字节码达到一定数量 (比如 100 条) 时，线程主动释放 GIL 全局锁并唤醒其他线程。其他等待 GIL 全局锁的线程取得锁后，将得到虚拟机控制权并开始执行。因此，虚拟机就像一颗软件 CPU ，Python 线程交替在虚拟机上执行

在 GIL 的束缚下，Python 虚拟机同一时刻只能执行一个线程。这是否意味着多线程完全无法优化程序性能呢？由于程序运行特征千差万别，这个问题得分情况讨论。开始之前，我们先来了解两种不同的运行特征：

程序在运行时，一般都处于两种状态：

• 可执行 状态，包括 Running 以及 Ready 两种情况，这时竞争处理器资源；
• 阻塞 状态，一般为等待 IO 处理，这时让出处理器资源；

根据程序分别处于 Running 以及 IO Blocked 两种状态的时间占比，可分为两种：

• 计算密集型 ，程序执行时大部分时间处于 Running 状态；
• IO 密集型 ，程序执行时大部分时间处于 IO Blocked 状态；

多线程的网络爬虫程序可以极大缩减时间

计算密集型的话 只能使用多进程

GIL是为了方便解释器开发人员而设置的, 尤其是在内存管理方面。 假设没有GIL, 那么两个线程在销毁同一个对象的时候, 就会出现悬空指针的问题。 于是python引用了GIL, 它是以字节码为单位的, 不管何时, 只有一个线程在执行字节码。 如果当前的某条字节码没有执行完, 那么是不会发生线程切换的, 因此这样就保证了以字节码为单位的安全性。 但是我们在写python代码时, 由于一行代码一般会对应多条字节码, 所以这个时候会再通过threading.Lock()进行加锁 所以: GIL是以字节码为单位的, 它保证了每条字节码在执行的时候不会被打断 而threading.Lock()是以代码为单位的, 它是保证python层面上的一行或多行代码在执行时不会被打断

发展几十年 有一些尝试 但是仍然不成熟 官方也没有

dict 的分析

哈希表的实现

key是不可以重复的 因为会按照key找对应的value

数据插入后，我们发现 dict 对象内存使用量保存不变。看来， dict 对象也有一种类似 list 对象的 预分配机制 。

哈希表结构决定了 dict 的删除操作也很快，平均时间复杂度也是 O(1)O(1) 。实际上， dict 插入、删除、查找的平均时间复杂度都是 ，O(1)O(1)最坏时间复杂度是 O(n)O(n) 。因此，哈希函数的选择就至关重要，一个好的哈希函数应该将键尽可能 均匀 地映射到哈希空间中，最大限度地避免 哈希冲突 。

dict 对象在 Python 内部由结构体 PyDictObject 表示， PyDictObject 在头文件 Include/dictobject.h 中定义

dict 对象理论上应该是一种变长对象，但 PyObject_HEAD 头部告诉我们， Python 其实把它作为普通对象实现。除了对象公共头部外， PyDictObject 还包括以下几个字段：

1. ma_used ，对象当前所保存的 键值对个数 ；
2. ma_version_tag ，对象当前 版本号 ，每次修改时更新；
3. ma_keys ，指向按键对象映射的 哈希表 结构；
4. ma_values ， 分离模式下指向由所有 值对象 组成的数组。

_dictkeysobject 结构体包含 dict 对象哈希表实现的所有秘密，结合注释可以解读其中的关键字段：

• dk_refcnt ，引用计数，跟 映射视图 的实现有关，有点类似对象引用计数；
• dk_size ，哈希表大小，必须是 2^n2
• n
• ，这样可将模运算优化成 按位与 运算；
• dk_lookup ， 哈希查找函数 指针，可根据 dict 当前状态选用最优函数版本；
• dk_usable ，键值对数组 可用个数 ；
• dk_nentries ，键值对数组 已用个数 ；
• dk_indices ，哈希表 起始地址 ，哈希表后紧接着 键值对数组 dk_entries 。

哈希表越密集，哈希冲突则越频繁，性能也就越差。因此，哈希表必须是一种 稀疏 的表结构，越稀疏则性能越好。由于 内存开销 的制约，哈希表不可能无限度稀疏，需要在时间和空间上进行权衡。实践经验表明，一个0.5-0.67 满的哈希表，性能较为理想——以相对合理的 内存 换取相对高效的 执行性能 。

为保证哈希表的稀疏程度，进而控制哈希冲突频率， Python 通过 USABLE_FRACTION 宏将哈希表内元素控制在 0.5-0.67 以内。USABLE_FRACTION 宏根据哈希表规模 n ，计算哈希表可存储元素个数，也就是 键值对数组 的长度。以长度为 8 的哈希表为例，最多可以保持 5 个键值对，超出则需要扩容。USABLE_FRACTION 是一个非常重要的宏定义，位于源文件 Objects/dictobject.c 中

• dict 是一种高效的关联式容器，每秒完成高达 200 多万次搜索操作；
• dict 内部由哈希表实现，哈希表的 稀疏 特性意味着昂贵的内存开销；
• 为优化内存使用， Python 将 dict 哈希表分为 哈希索引 和 键值对 两个数组来实现；
• 哈希表在 0.5-0.67 满时，性能较为理想，较好地平衡了 内存开销 与 搜索效率 ；

解决哈希冲突的常用方法有两种：

分离链接法 ( separate chaining ) ；
开放地址法 ( open addressing )；

​