# 分布式函数调度框架
## 环境及功能介绍
* python 3.9
* funboost框架
```
pip3 install -i http://mirrors.al2iyun.com/pypi/simple/  funboost --upgrade --trusted-host mirrors.aliyun.com
```



### 介绍
 分布式函数调度框架，支持5种并发模式，20+消息中间件，20种任务控制功能。
 
 用途概念就是常规经典的：生产者+消息队列中间件+消费者  编程思想
### 5种并发模式
* threading（使用的是可变线程池，可以智能自动缩小和扩大线程数量）
* gevent
* evenlet
* asyncio(框架可以直接支持asynic定义的协程函数作为任务，celery不支持)
* single_thread
### 20多种中间件
```
RABBITMQ_AMQPSTORM = 0  # 使用 amqpstorm 包操作rabbitmq  作为 分布式消息队列，支持消费确认.推荐这个。

RABBITMQ_RABBITPY = 1  # 使用 rabbitpy 包操作rabbitmq  作为 分布式消息队列，支持消费确认。

REDIS = 2  # 使用 redis 的 list结构，brpop 作为分布式消息队列。随意重启和关闭会丢失大量消息任务，不支持消费确认。

LOCAL_PYTHON_QUEUE = 3  # 使用python queue.Queue实现的基于当前python进程的消息队列，不支持跨进程 跨脚本 跨机器共享任务，不支持持久化，适合一次性短期简单任务。

RABBITMQ_PIKA = 4  # 使用pika包操作rabbitmq  作为 分布式消息队列。

MONGOMQ = 5  # 使用mongo的表中的行模拟的 作为分布式消息队列，支持消费确认。

PERSISTQUEUE = 6  # 使用基于sqlute3模拟消息队列，支持消费确认和持久化，但不支持跨机器共享任务，可以基于本机单机跨脚本和跨进程共享任务，好处是不需要安装中间件。

NSQ = 7  # 基于nsq作为分布式消息队列，支持消费确认。

KAFKA = 8  # 基于kafka作为分布式消息队列，建议使用BrokerEnum.CONFLUENT_KAFKA。

REDIS_ACK_ABLE = 9  # 基于redis的 list + 临时unack的set队列，采用了 lua脚本操持了取任务和加到pengding为原子性，随意重启和掉线不会丢失任务。
                  

SQLACHEMY = 10  # 基于SQLACHEMY 的连接作为分布式消息队列中间件支持持久化和消费确认。支持mysql oracle sqlserver等5种数据库。

ROCKETMQ = 11  # 基于 rocketmq 作为分布式消息队列，这个中间件必须在linux下运行，win不支持。

REDIS_STREAM = 12  # 基于redis 5.0 版本以后，使用 stream 数据结构作为分布式消息队列，支持消费确认和持久化和分组消费，是redis官方推荐的消息队列形式，比list结构更适合。

ZEROMQ = 13  # 基于zeromq作为分布式消息队列，不需要安装中间件，可以支持跨机器但不支持持久化。

RedisBrpopLpush = 14  # 基于redis的list结构但是采用brpoplpush 双队列形式，和 redis_ack_able的实现差不多，实现上采用了原生命令就不需要lua脚本来实现取出和加入unack了。

"""
操作 kombu 包，这个包也是celery的中间件依赖包，这个包可以操作10种中间件(例如rabbitmq redis)，但没包括分布式函数调度框架的kafka nsq zeromq 等。
同时 kombu 包的性能非常差，可以用原生redis的lpush和kombu的publish测试发布，使用brpop 和 kombu 的 drain_events测试消费，对比差距相差了5到10倍。
由于性能差，除非是分布式函数调度框架没实现的中间件才选kombu方式(例如kombu支持亚马逊队列  qpid pyro 队列)，否则强烈建议使用此框架的操作中间件方式而不是使用kombu。
"""
KOMBU = 15

"""基于confluent-kafka包，包的性能比kafka-python提升10倍。同时应对反复随意重启部署消费代码的场景，此消费者实现至少消费一次，第8种BrokerEnum.KAFKA是最多消费一次。"""
CONFLUENT_KAFKA = 16

""" 基于emq作为中间件的。这个和上面的中间件有很大不同，服务端不存储消息。所以不能先发布几十万个消息，然后再启动消费。mqtt优点是web前后端能交互，
前端不能操作redis rabbitmq kafka，但很方便操作mqtt。这种使用场景是高实时的互联网接口。
"""
MQTT = 17

HTTPSQS = 18  # 基于httpsqs的

PULSAR = 20   # 下一代分布式消息系统。5年后会同时取代rabbitmq和kafka。

UDP = 21  # 基于socket udp 实现的。小规模使用不支持持久化，好处是不用安装软件。

TCP = 22  # 基于socket tcp 实现的。小规模使用不支持持久化，好处是不用安装软件。

HTTP = 23 # 基于http实现的，小规模使用不支持持久化，好处是不用安装软件。

NATS = 24  # 高性能中间件nats,中间件服务端性能很好,。

TXT_FILE = 25 # 磁盘txt文件作为消息队列，支持单机持久化，不支持多机分布式

PEEWEE = 26 # peewee包操作mysql，使用peewee包操作mysql表模拟消息队列

REDIS_PUBSUB = 27 #  使用redis pubsub实现的广播，每个消费者都会消费同一条消息。
```

### 20种控制功能
```
分布式：
支持数十种最负盛名的消息中间件.(除了常规mq，还包括用不同形式的如 数据库 磁盘文件 redis等来模拟消息队列)

并发：
支持threading gevent eventlet asyncio 单线程 5种并发模式 叠加 多进程。
多进程不是和前面四种模式平行的，是叠加的，例如可以是 多进程 + 协程，多进程 + 多线程。

控频限流：
例如十分精确的指定1秒钟运行30次函数或者0.02次函数（无论函数需要随机运行多久时间，都能精确控制到指定的消费频率；

分布式控频限流：
例如一个脚本反复启动多次或者多台机器多个容器在运行，如果要严格控制总的qps，能够支持分布式控频限流。

任务持久化：
消息队列中间件天然支持

断点接续运行：
无惧反复重启代码，造成任务丢失。消息队列的持久化 + 消费确认机制 做到不丢失一个消息
(此框架很重视消息的万无一失，就是执行函数的机器支持在任何时候随时肆无忌惮反复粗暴拉电闸断电，或者强制硬关机，
或者直接用锄头把执行函数代码的机器砸掉，只要不是暴力破坏安装了消息队列中间件的机器就行，消息就万无一失，
现在很多人做的简单redis list消息队列，以为就叫做分布式断点接续，那是不正确的，因为这种如果把消息从reidis brpop取出来后，
如果消息正在被执行，粗暴的kill -9脚本或者直接强制关机，那么正在运行的消息就丢失了，如果是多线程同时并发运行很多消息，粗暴重启
会丢失几百个大量消息，这种简单的redis list根本就不能叫做安全的断点续传。
分布式函数调度框架的消费确认机制，保证函数运行完了才确认消费，正在运行突然强制关闭进程不会丢失一个消息，
下次启动还会消费或者被别的机器消费。
此框架的消息万无一失特性，不仅支持rabbbitmq因为原生支持，也支持redis，框架对redis的实现机制是因为客户端加了一层保障)。

定时：
可以按时间间隔、按指定时间执行一次、按指定时间执行多次，使用的是apscheduler包的方式。

延时任务：
 例如规定任务发布后，延迟60秒执行，或者规定18点执行。这个概念和定时任务有一些不同。
	  
指定时间不运行：
例如，有些任务你不想在白天运行，可以只在晚上的时间段运行

消费确认：
这是最为重要的一项功能之一，有了这才能肆无忌惮的任性反复重启代码也不会丢失一个任务。
（常规的手写 redis.lpush + redis.blpop,然后并发的运行取出来的消息，随意关闭重启代码瞬间会丢失大量任务，
那种有限的 断点接续 完全不可靠，根本不敢随意重启代码）

立即重试指定次数：
当函数运行出错，会立即重试指定的次数，达到最大次重试数后就确认消费了

重新入队：
在消费函数内部主动抛出一个特定类型的异常ExceptionForRequeue后，消息重新返回消息队列

超时杀死：
例如在函数运行时间超过10秒时候，将此运行中的函数kill

计算消费次数速度：
实时计算单个进程1分钟的消费次数，在日志中显示；当开启函数状态持久化后可在web页面查看消费次数

预估消费时间：
根据前1分钟的消费次数，按照队列剩余的消息数量来估算剩余的所需时间

函数运行日志记录：
使用自己设计开发的 控制台五彩日志（根据日志严重级别显示成五种颜色；使用了可跳转点击日志模板）
+ 多进程安全切片的文件日志 + 可选的kafka elastic日志
		   
任务过滤：
例如求和的add函数，已经计算了1 + 2,再次发布1 + 2的任务到消息中间件，可以让框架跳过执行此任务。
任务过滤的原理是使用的是函数入参判断是否是已近执行过来进行过滤。

任务过滤有效期缓存：
例如查询深圳明天的天气，可以设置任务过滤缓存30分钟，30分钟内查询过深圳的天气，则不再查询。
30分钟以外无论是否查询过深圳明天的天气，则执行查询。

任务过期丢弃：
例如消息是15秒之前发布的，可以让框架丢弃此消息不执行，防止消息堆积,
在消息可靠性要求不高但实时性要求高的高并发互联网接口中使用
		
函数状态和结果持久化：
可以分别选择函数状态和函数结果持久化到mongodb，使用的是短时间内的离散mongo任务自动聚合成批量
任务后批量插入，尽可能的减少了插入次数
			  
消费状态实时可视化：
在页面上按时间倒序实时刷新函数消费状态，包括是否成功 出错的异常类型和异常提示 
重试运行次数 执行函数的机器名字+进程id+python脚本名字 函数入参 函数结果 函数运行消耗时间等
			 
消费次数和速度生成统计表可视化：
生成echarts统计图，主要是统计最近60秒每秒的消费次数、最近60分钟每分钟的消费次数
最近24小时每小时的消费次数、最近10天每天的消费次数
						
rpc：
生产端（或叫发布端）获取消费结果。各个发布端对消费结果进行不同步骤的后续处理更灵活，而不是让消费端对消息的处理一干到底。

远程服务器部署消费函数：
代码里面 task_fun.fabric_deploy('192.168.6.133', 22, 'xiaomin', '123456', process_num=2) 只需要这样就可以自动将函数部署在远程机器运行，
无需任何额外操作，不需要借助阿里云codepipeline发版工具 和 任何运维发版管理工具，就能轻松将函数运行在多台远程机器。task_fun指的是被@boost装饰的函数

暂停消费：
支持从python解释器外部/远程机器 ，控制暂停消息消费和继续消费。
```




TypeError: cannot set 'is_timeout' attribute of immutable type 'TimeoutError'的错误，后来查询CSDN后运行
```
pip install https://github.com/eventlet/eventlet/archive/master.zip
```
解决了问题.

funboost框架的学习

backend-ToC

面试官，请帮我回答一下以下问题

[印象笔记-第6周总结](https://app.yinxiang.com/fx/640c0d15-bb3f-40d8-8b66-04ec9ddb50cf)

第6周总结

UUID 通常用作数据库表主键。它们易于生成，易于在分布式系统之间共享并保证唯一性。

考虑到 UUID 的大小，这是否是一个正确的选择值得怀疑，但通常这不是由我们决定的。

本文的重点不是“UUID 是否是键的正确格式”，而是如何有效地使用 UUID 作为 PostgreSQL 的主键。



## UUID 的 Postgres 数据类型

UUID 可以被视为一个字符串，并且可能很容易将它们存储为字符串。 Postgres 具有用于存储字符串的灵活数据类型： text ，并且通常用作存储 UUID 值的主键。

它是正确的数据类型吗？当然不。

Postgres 有一个专用于 UUID 的数据类型： uuid 。 UUID 是 128 位数据类型，因此存储单个值需要 16 个字节。 text 数据类型有 1 或 4 个字节的开销加上存储实际的字符串。

这些差异在小表中并不那么重要，但一旦开始存储数十万或数百万行，就会成为问题。

我进行了一个实验，看看实践中有何不同。有两个表只有一列 - id 作为主键。第一个表使用 text ，第二个表使用 uuid ：
```sql
create table bank_transfer(
    id text primary key
);

create table bank_transfer_uuid(
    id uuid primary key
);
```

我没有指定主键索引的类型，因此 Postgres 使用默认的 B 树。

向每个表插入 10 000 000 行:

我运行查询来查找表大小和索引大小：
```sql
select 
    relname as "table", 
    indexrelname as "index",
    pg_size_pretty(pg_relation_size(relid)) "table size",
    pg_size_pretty(pg_relation_size(indexrelid)) "index size"
from 
    pg_stat_all_indexes
where 
    relname not like 'pg%';

+------------------+-----------------------+----------+----------+
|table             |index                  |table size|index size|
+------------------+-----------------------+----------+----------+
|bank_transfer_uuid|bank_transfer_uuid_pkey|422 MB    |394 MB    |
|bank_transfer     |bank_transfer_pkey     |651 MB    |730 MB    |
+------------------+-----------------------+----------+----------+
```

使用 text 的表增大了 54%，索引大小增大了 85%。这也反映在 Postgres 用于存储这些表和索引的页数上：

```sql
select relname, relpages from pg_class where relname like 'bank_transfer%';

+-----------------------+--------+
|relname                |relpages|
+-----------------------+--------+
|bank_transfer          |83334   |
|bank_transfer_pkey     |85498   |
|bank_transfer_uuid     |54055   |
|bank_transfer_uuid_pkey|50463   |
+-----------------------+--------+
```

更大的表、索引和更多的表意味着 Postgres 必须执行插入新行和获取行的工作 - 特别是当索引大小大于可用 RAM 内存时，Postgres 必须从磁盘加载索引。

## UUID 和 B 树索引

随机 UUID 不太适合 B 树索引 - 并且 B 树索引是主键唯一可用的索引类型。

B 树索引最适合处理有序值 - 例如自动递增列或时间排序列。

UUID - 尽管看起来总是相似 - 有多种变体。 Java 的 UUID.randomUUID() - 返回 UUID v4 - 这是一个伪随机值。对我们来说，更有趣的是 UUID v7 - 它生成按时间排序的值。这意味着每生成一个新的UUID v7，它就有一个更大的值。这使得它非常适合 B 树索引。

python如何生成uuid v7:

```python
import uuid

def generate_uuid_v7():
    return uuid.uuid7()

if __name__ == "__main__":
    # 生成 UUID v7
    uuid_v7 = generate_uuid_v7()
    print(f"生成的 UUID v7: {uuid_v7}")

```

运行上述代码后，输出可能类似于：

```
生成的 UUID v7: 7b2fa7a6-0dd8-7e43-b27e-9e735f8d5e60
```


## UUID v7 如何影响 INSERT 性能

我创建了另一个表，与 bank_transfer_uuid 完全相同，但它将仅存储使用上述库生成的 UUID v7：

```sql
create table bank_transfer_uuid_v7(
   id uuid primary key
);
```

然后，我运行 10 轮，向每个表插入 10000 行，并测量需要多长时间：

结果看起来有点随机，尤其是在比较具有常规 text 列和 uuid v4 的表的时间时：

```sql
+-------+-------+---------+
| text  | uuid  | uuid v7 |
+-------+-------+---------+
| 7428  | 8584  | 3398    |
| 5611  | 4966  | 3654    |
| 13849 | 10398 | 3771    |
| 6585  | 7624  | 3679    |
| 6131  | 5142  | 3861    |
| 6199  | 10336 | 3722    |
| 6764  | 6039  | 3644    |
| 9053  | 5515  | 3621    |
| 6134  | 5367  | 3706    |
| 11058 | 5551  | 3850    |
+-------+-------+---------+
```

但我们可以清楚地看到，插入 UUID v7 比插入常规 UUID v4 快约 2 倍。



## 概括
正如一开始提到的 - 由于 UUID 长度 - 即使进行了所有这些优化，它也不是主键的最佳类型。如果您可以选择，请查看由 Vlad Mihalcea 维护的 TSID。

但如果您必须或出于某种原因想要使用 UUID，请考虑我提到的优化。另请记住，此类优化对于大型数据集会产生影响。如果您存储数百甚至数千行，并且流量较低，您可能不会看到应用程序性能有任何差异。但是，如果您有可能拥有大型数据集或大流量 - 最好从一开始就这样做，因为更改主键可能是一个相当大的挑战。




[深入阅读](https://vladmihalcea.com/uuid-database-primary-key/)