from rq import Queue
from redis import Redis  # 配置Redis連接（生產環境需加密碼/超時）
from my_tasks import send_email
# 連接Redis並指定隊列
redis_conn = Redis(host="localhost", port=6379, password="your-pwd", decode_responses=False)
q = Queue(name="email", connection=redis_conn)  # 顯式指定隊列名（生產最佳實踐）
# 入隊任務：函數+參數
job = q.enqueue(send_email, "user@example.com", subject="Welcome")

# 監聽單個隊列（email隊列）
rq worker email --connection redis://:your-pwd@localhost:6379/0
# 監聽多個隊列（優先級：high > default > low）
rq worker high default low
# 後台運行（生產環境用supervisor守護）
nohup rq worker email > rq-worker.log 2>&1 &

def worker_loop(queue_names, redis_conn):
while True:
# 1. 阻塞獲取任務（優先級高的隊列先查）
# BLPOP是原子操作：多個Worker不會搶到同一個任務
queue_key, job_id = redis_conn.blpop([f"rq:queue:{name}" for name in queue_names], timeout=0)
# 2. 標記任務為"執行中"（防止超時後被重複執行）
redis_conn.hset(f"rq:job:{job_id}", "status", "started")
redis_conn.hset(f"rq:job:{job_id}", "worker_name", self.name)
# 3. 執行任務（獲取元信息→反序列化→調用函數）
job_meta = redis_conn.hgetall(f"rq:job:{job_id}")
func = import_string(job_meta["func_name"])  # 導入任務函數
args = pickle.loads(job_meta["args"])        # 反序列化參數
try:
result = func(*args)  # 執行核心邏輯
# 4. 執行成功：存結果+設狀態
redis_conn.setex(
f"rq:job:{job_id}:result",
job_meta["result_ttl"],
pickle.dumps(result)
)
redis_conn.hset(f"rq:job:{job_id}", "status", "finished")
except Exception as e:
# 5. 執行失敗：移到失敗隊列/重試隊列
if job_meta.get("retry_count") < job_meta.get("max_retry"):
# 延遲重試：加入Sorted Set（score=下次執行時間）
next_retry_ts = time.time() + get_retry_interval(job_meta["retry_count"])
redis_conn.zadd(
f"rq:queue:{queue_name}:retry",
{job_id: next_retry_ts}
)
else:
# 徹底失敗：移到failed隊列
redis_conn.rpush("rq:queue:failed", job_id)
redis_conn.hset(f"rq:job:{job_id}", "status", "failed")
redis_conn.hset(f"rq:job:{job_id}", "error", str(e))

from rq import Queue
from redis import Redis
# 1. 建立Redis連接池（複用連接，避免頻繁創建）
redis_pool = Redis(
host="redis-host",
port=6379,
password="your-strong-pwd",
db=0,
socket_timeout=5,
max_connections=100,
decode_responses=False
).connection_pool
# 2. 按任務類型分隊列（優先級+職責隔離）
high_queue = Queue("high", connection=Redis(connection_pool=redis_pool))  # 核心任務
email_queue = Queue("email", connection=Redis(connection_pool=redis_pool))  # 郵件任務
low_queue = Queue("low", connection=Redis(connection_pool=redis_pool))  # 非核心任務
# 3. 關鍵任務必須指定超時和重試
from rq import Retry
high_queue.enqueue(
process_payment,  # 核心支付任務
user_id=123,
amount=99.9,
job_timeout=60,  # 短超時（核心任務快失敗快重試）
retry=Retry(max=3, interval=[10, 30, 60]),  # 指數退避重試
result_ttl=3600  # 結果保留1小時（供後續對賬）
)

# 示例：supervisor配置文件（/etc/supervisor/conf.d/rq-worker.conf）
[program:rq-high-worker]
command=rq worker high --connection redis://:your-pwd@redis-host:6379/0
directory=/path/to/your/project
user=appuser  # 非root用户運行
autostart=true
autorestart=true  # 進程崩潰自動重啓
redirect_stderr=true
stdout_logfile=/var/log/rq/high-worker.log
stdout_logfile_maxbytes=10MB
stdout_logfile_backups=10
stopasgroup=true  # 停止Worker時同時停止子進程
killasgroup=true

import json
from rq.serializers import Serializer
# 自定義JSON序列化器
class JSONSerializer(Serializer):
@staticmethod
def dumps(obj):
return json.dumps(obj).encode('utf-8')
@staticmethod
def loads(obj):
return json.loads(obj.decode('utf-8'))
# 隊列使用自定義序列化器
q = Queue("safe-queue", connection=redis_conn, serializer=JSONSerializer)