如何快速创建一个拥有异步任务队列集群的 Rest Api

时间：2022-02-25 01:08:46

如何快速创建一个拥有异步任务队列集群的 Rest Api

异步任务是 Web 后端开发中最常见的需求，非常适合多任务、高并发的场景。本文分享如何使用 docker-compose、FastAPI、rq 来快速创建一个包含异步任务队列集群的 REST API，后端执行任务的节点可以随意扩展。

系统的架构图：

如何快速创建一个拥有异步任务队列集群的 Rest Api

上图中的每一个方框都可以理解为一个服务器。

用户请求 api， api 将任务放入 redis 队列，worker 自动去 redis 队列取出任务并执行，worker 节点可以任意水平扩展。

接下来，我们来实现这一架构的 demo，你可以看到 docker 的强大和方便之处。

1、先创建一个虚拟环境，安装依赖

依赖 fastapi，redis，rq 库，安装后生成一个 requirements.txt 文件

  1. mkdir myproject
  2. python3 -m venv env
  3. source env/bin/activate
  4. pip install rq
  5. pip install fastapi
  6. pip install redis
  7. pip freeze > requirements.txt

2、编码实现 REST API、Worker

REST 是一种风格，这里不是重点，我们使用 FastAPI 来快速创建一个接口，新建一个 api.py 的文件，内容如下：

  1. from fastapi import FastAPI
  2. from redis import Redis
  3. from rq import Queue
  4. from worker import send_captcha
  5. app = FastAPI()
  7. # 需要注意，这里的 host 是主机名，在 docker 中就是服务名，后面的 docker-compose.ymal 中的服务名称也要是这个
  8. redis_conn = Redis(host='myproj_redis', port=6379, db=0)
  10. # 定义一个队列，名称是 my_queue
  11. q = Queue('my_queue', connection=redis_conn)
  13. @app.get('/hello')
  14. def hello():
  15. """Test endpoint"""
  16. return {'hello': 'world'}
  18. # Rest API 示例
  19. @app.post('/send_captcha/{phone_number}', status_code=201)
  20. def addTask(phone_number: str):
  21. """
  22. Adds tasks to worker queue.
  23. Expects body as dictionary matching the Group class.
  25. """
  26. job = q.enqueue(send_captcha, phone_number)
  28. return {'job': "tasks add done."}

这里的 send_captcha 函数就是一个异步任务，从 worker.py 中导入，worker.py 的内容如下：

  1. import time
  3. def send_captcha(phone_number):
  4. """
  5. 模拟一个耗时的异步任务
  6. """
  7. print(f'{time.strftime("%T")} 准备发送手机验证码') # in place of actual logging
  8. print(f'{time.strftime("%T")} 生成随机验证码并存入 redis，设置 5 分钟过期时间')
  9. time.sleep(5) # simulate long running task
  10. print(f'{time.strftime("%T")} {phone_number}发送完成')
  11. return { phone_number: 'task complete'}

return { phone_number: 'task complete'}

3、构建 Dokcer 镜像

现在的目标是实现一个拥有两个执行节点的集群。我们需要启动 4 个容器来完成一个集群部署：

  • 容器1：运行 FastAPI app
  • 容器2：运行 Redis 服务
  • 容器3：运行 worker 1 服务
  • 容器4：运行 worker 2 服务

其中容器 1、3、4 都是 Python 应用，可以共用一个 Python 镜像。

为了方便调试，我们可以让 1、3、4 容器共享我们的本地路径，这样改了代码就不需要重新构建镜像，比较方便。

创建一个包含依赖的 Python 镜像

现在我们来创建一个包含前文 requirements.txt 依赖的 Python 镜像，编写 Dockerfile，内容如下：

  1. FROM python:3.8-alpine
  2. RUN adduser -D myproj
  3. WORKDIR /home/myproj
  4. COPY requirements.txt requirements.txt
  5. RUN pip install -r requirements.txt
  6. RUN chown -R myproj:myproj ./
  7. USER myproj
  8. CMD uvicorn api:app --host 0.0.0.0 --port 5057

内容说明：

FROM python:3.8-alpine

指定使用 python:3.8-alpine，这个容器已经预装了 Python3.8，可以在命令行执行 docker search python 看看有哪些 Python 镜像。

RUN adduser -D myproj

添加一个用户 myproj，这一步的主要目的是为了生成目录 /home/myproj

WORKDIR /home/myproj

设置程序的执行路径为 /home/myproj

COPY requirements.txt requirements.txt

复制当前路径下的 requirements.txt 到容器的 /home/myproj，这里没有复制 .py 文件是因为后面我们启动容器的时候会共享本地路径，不需要再复制了，生产部署时最好复制到窗口内部，这样容器就不会依赖本机。

RUN pip install -r requirements.txt

在容器中安装依赖

RUN chown -R myproj:myproj ./

将 /home/myproj 路径下的文件的拥有者和所属组改为 myproj，这一步为了使用 myproj 用户来启动 fastapi 服务，生产环境通常用 root 用户启动，也就不需要这个指令了。

USER myproj

切换到 myproj 用户

CMD uvicorn api:app --host 0.0.0.0 --port 5057

容器启动后执行的命令，服务端口为 5057

更多的 Dockerfile 语法请参考官方文档，这里仅是简要说明。

现在 Dockerfile 所在的目录执行下面的命令构建一个镜像：

  1. docker build -t myproject:latest .

创建完成后，可以使用 docker images 来查看：

  1. ❯ docker images | grep myproj
  2. myproject

4、启动集群

这里使用 Docker Compose 来启动 4 个容器，为什么用 Docker Compose 呢?因为方便，如果不用的话，需要手动一个容器一个容器启动。

Docker Compose 会读取一个 yaml 格式的配置文件，依据配置文件来启动容器，各容器共享同一网络。还记得 api.py 中使用的 Redis 主机名吗，这里就需要将 redis 服务名设置为那个主机名。

编写一个 docker-compose.yml 内容如下：

  1. version: '3'
  3. services:
  4. myproj_redis:
  5. image: redis:4.0-alpine
  6. ports:
  7. - "6379:6379"
  8. volumes:
  9. - ./redis:/data
  11. myproj_api:
  12. image: myproject:latest
  13. command: uvicorn api:app --host 0.0.0.0 --port 5057
  14. ports:
  15. - "5057:5057"
  16. volumes:
  17. - ./:/home/myproj
  19. myproj_worker1:
  20. image: myproject:latest
  21. command: rq worker --url redis://myproj_redis:6379 my_queue
  22. volumes:
  23. - ./:/home/myproj
  25. myproj_worker2:
  26. image: myproject:latest
  27. command: rq worker --url redis://myproj_redis:6379 my_queue
  28. volumes:
  29. - ./:/home/myproj

第一个容器是 myproj_redis，运行着 redis 服务， redis 的数据通过 volumes 方式保存在本地，因此需要在本地创建一个 redis 目录，来映射容器内部的 /data 目录。

第二个容器就是 fastapi 服务，端口 5057，使用本地路径映射为 /home/myproj

第三个容器和第四个容器是 worker 节点，虽然也映射了本地路径，但它仅使用 worker.py 文件。当任务太多时，worker 节点可以扩展，解决负载压力，

最终的目录是这样：

如何快速创建一个拥有异步任务队列集群的 Rest Api

执行 docker compose 命令启动 4 个容器:

  1. docker compose -f docker-compose.yml up

可以看到 4 个服务均启动并正常打印了日志输出。

如何快速创建一个拥有异步任务队列集群的 Rest Api

4、测试

现在来测试一下，左边的窗口，我使用 Python 快速发送了 3 个 post 请求:

  1. import subprocess
  2. for i in range(3):
  3. subprocess.run("curl -v -X POST 'http://localhost:5057/send_captcha/18012345678'",shell = True)

从右边窗口的日志输出可以看出 worker1 和 worker2 都执行了任务，其中 worker1 执行了 2 个，worker2 执行了 1 个。

如何快速创建一个拥有异步任务队列集群的 Rest Api

查看完整代码请点击「阅读原文」

最后的话

本文分享了如何使用 Dockerfile 构建一个镜像，使用 Docker Compose 管理一个容器集群，以此为基础实现了一个具有异步任务队列集群的 REST API，抛砖引玉，关于 Dockerfile、docker-compose 的详细用法，还请参考 Docker 官方文档

原文链接：https://mp.weixin.qq.com/s/Agi74bM5bNU9gr_g0TbITQ

标签：

相关文章