[经验] python下的消息队列选择-rq

7月 19, 2018 0条评论 16,044次阅读 6人点赞

背景

某个web api项目需要将api的每次调用情况都记录下来，统一收集之后存入数据库，然后每个小时进行汇总与统计。这样方便业务部门进行数据分析，逻辑流程如下

用户请求 -> web api(记录请求) -> 请求返回数据 并记录本次请求 -> 业务人员汇总数据

1 2	用户请求 -> web api(记录请求) -> 请求返回数据并记录本次请求 -> 业务人员汇总数据

从上面的看出，api业务访问数据和记录本次的访问行为其实可以完全分开，对于用户的视角，只会关注本次的请求是否OK，而对于运营人员，更多的关注是本次访问的行为结果，所以，在逻辑上我们需要将两个动作分开，如果按照早期的做法，我们大可以做如下的动作

1.http 请求 返回数据
2.记录本次http请求成功与否到数据库中

1.http 请求返回数据

2.记录本次http请求成功与否到数据库中

每次的请求都绑定了两个动作，如果并发多了(对于http应用来说是常态)，对于日志数据库来说就是一个压力了，当然，我们可以做读写分离了来减轻数据库层面的压力，但是这样的耦合度太高了，对于用户来说api请求时间也有消耗，所以这种类似的异步操作，我们可以使用消息队列完成，本次，我主要从三种常用消息队列中间件来介绍

准备

python
docker-compose
python-rq

安装redis环境

使用docker-compose 起一个redis 环境

创建一个redis的文件夹，在里面写入名字docker-compose.yml 的文件，内容如下

version: '2'

services:
  redis:
    image: redis
        command: redis-server --requirepass "pythonrq"
    restart: always
    ports:
      - "6666:6379"

version: '2'

services:

redis:

image: redis

command: redis-server --requirepass "pythonrq"

restart: always

ports:

- "6666:6379"

使用命令docker-compose up -d

然后使用命令docker-compose ps我们可以看到如下的输出

    Name                   Command               State           Ports
-------------------------------------------------------------------------------
redis_redis_1   docker-entrypoint.sh redis ...   Up      0.0.0.0:6666->6379/tcp

Name Command State Ports

-------------------------------------------------------------------------------

redis_redis_1 docker-entrypoint.sh redis ... Up 0.0.0.0:6666->6379/tcp

这样我们已经使用docker-compose 起了一个redis服务了

安装python-rq

pip install rq

1 2	pip install rq

OK，我们准备一个简单的web api 项目，本次选用的web框架为vibora

准备web框架

pip install vibora

1 2	pip install vibora

核心代码

app_factory.py 代码

# -*- coding: utf-8 -*-
from vibora import Vibora
from vibora.router import RouterStrategy
from src.views.api import bp_api

from src.settings import app_config
from src.log.config import logger
from redis import StrictRedis
from rq import Queue


def create_app():
    app = Vibora(router_strategy=RouterStrategy.CLONE)
    # app.configure_static_files()
    app.add_blueprint(bp_api)
    app.logger = logger

    job_queue = Queue(connection=StrictRedis.from_url(app_config.redis_conn))
    app.components.add(job_queue)

    return app

# -*- coding: utf-8 -*-

from vibora import Vibora

from vibora.router import RouterStrategy

from src.views.api import bp_api

from src.settings import app_config

from src.log.config import logger

from redis import StrictRedis

from rq import Queue

def create_app():

app = Vibora(router_strategy=RouterStrategy.CLONE)

# app.configure_static_files()

app.add_blueprint(bp_api)

app.logger = logger

job_queue = Queue(connection=StrictRedis.from_url(app_config.redis_conn))

app.components.add(job_queue)

return app

我们在创建web app的时候，同时创建了一个组件，加入到了整个app runtime 中

api.py文件

# -*- coding: utf-8 -*-
import time

from vibora.responses import JsonResponse
from vibora.blueprints import Blueprint
from rq import Queue

from src.log.config import logger
from src.job.word import count_words

bp_api = Blueprint()


@bp_api.route('/time')
async def get_time(job_queue: Queue):
    logger.info('get user request')

    job_queue.enqueue(count_words, str(time.time()))
    return JsonResponse({'time': str(time.time())})