python分布式框架rq的使用

RedisQueue是一款輕量級的分布式異步任務隊列調度框架，基於redis數據庫作為broker，生產端將任務job存儲到redis數據庫中，消費端監聽隊列並取出任務執行。

1.基礎架構

rq框架使用前需要安裝rq庫，使用pip安裝即可：pip install rq

1.1 生產者

生產者將任務發送到指定redis的指定隊列中，job.py：

import requests
from rq import Queue

# 具體業務執行代碼
def count_words_at_url(url):
    resp = requests.get(url)
    return len(resp.text.split())

conn = redis.Redis(host='127.0.0.1', password='123456', port=8888, db=10) # 指定redis數據庫
q = Queue("high", connection=conn)                                        # 指定隊列

job_urls = ['http://baidu.com', 'http://qq.com', 'http://sohu.com']
for url in job_urls:
    q.enqueue_call(count_words_at_url, args=(url,))                       # 發送任務

1.2 消費者

消費者采用多進程架構，監聽指定隊列從中取出任務消息，並調用執行業務處理函數，worker.py：

import redis
from multiprocessing import Pool
from rq import Worker, Queue, Connection

conn = redis.Redis(host='127.0.0.1', password='123456', port=8888, db=10) # 同一個redis連接

# worker邏輯，無需修改
def worker(listen):
    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

if __name__ == "__main__":
    listen = ['high', 'default', 'low']            # 監聽的隊列，可自定義修改
    try:
        cpu_num = 4
        p = Pool(cpu_num)
        for i in range(cpu_num):
            p.apply_async(worker, args=(listen,))  # 開啟worker
        p.close()
        p.join()
    except Exception as e:
        print(e)

2.基於redis集群

若是基於redis集群，可以用如下方式獲取redis連接：

from rq.cli.helpers import get_redis_from_config

settings = {'SENTINEL': {
	                     'INSTANCES': [('redis1.cloud.bz', 8080), 
                                       ('redis2.cloud.bz', 8080),
				                       ('redis3.cloud.bz', 8080)],
	                     'SOCKET_TIMEOUT': None, 
                         'PASSWORD': '', 
                         'DB': 10, 
                         'MASTER_NAME': 'sentinel-127.0.0.1-8080'
                        }
           }

conn = get_redis_from_config(settings)