python 3.10的Callable导入错误

[Ninggirsu]

使用python3.10版本运行测试案例报错

问题已解决

报错情况：

  File "C:\Users\90957\Documents\2022-1\fastapitest\test.py", line 2, in <module>
    from funboost import boost, BrokerEnum
  File "C:\software\Anaconda\envs\funboost\lib\site-packages\funboost\__init__.py", line 12, in <module>
    from funboost.consumers.base_consumer import (ExceptionForRequeue, ExceptionForRetry,
  File "C:\software\Anaconda\envs\funboost\lib\site-packages\funboost\consumers\base_consumer.py", line 25, in <module>
    from collections import Callable
ImportError: cannot import name 'Callable' from 'collections' (C:\software\Anaconda\envs\funboost\lib\collections\__init__.py)

原因：Deprecated since version 3.9: collections.abc.Callable Callable在python3.9已弃用
解决方案：使用低于3.9的python

出错的环境的配置完整命令：

conda create -n env_name python
conda activate env_name 
pip install funboost --upgrade

使用的代码

import time
from funboost import boost, BrokerEnum


@boost("task_queue_name1", qps=5, broker_kind=BrokerEnum.PERSISTQUEUE)  # 入参包括20种，运行控制方式非常多，想得到的控制都会有。
def task_fun(x, y):
    print(f'{x} + {y} = {x + y}')
    time.sleep(3)  # 框架会自动并发绕开这个阻塞，无论函数内部随机耗时多久都能自动调节并发达到每秒运行 5 次 这个 task_fun 函数的目的。


if __name__ == "__main__":
    for i in range(100):
        task_fun.push(i, y=i * 2)  # 发布者发布任务
    task_fun.consume()  # 消费者启动循环调度并发消费任务

[ydf0509]

from typing import Callable 改成这个了，下次发布版本