Python结合Redis Pub/Sub实现轻量级作业调度系统

系统原理与架构

基于Redis内存特性与丰富的数据类型，利用其原生支持的消息发布/订阅（Pub/Sub）功能，可以搭建一个低延迟的轻量级作业调度框架。系统分为主控节点（Master）与工作节点（Slave）：Master负责生成任务ID并向指定频道发布调度指令，同时订阅结果频道收集反馈；Slave监听对应分发频道，执行模拟任务后将执行状态回传至共享的结果频道。该模式无需外部消息中间件，适合本地测试或微服务间的简单异步通信。

环境准备

确保已安装Python 3.x，并通过 pip install redis 安装依赖库。本地需运行一个标准的Redis服务（默认地址 localhost:6379）。脚本中将直接使用 redis.StrictRedis 连接实例，无需额外配置集群，单节点即可满足本示例的跑通需求。

第1步：编写主控节点（Master）代码

目的：初始化调度逻辑，启动任务派发线程与结果处理线程。入口处直接实例化 MyMaster 类并调用 start()。
动作参数：设定Redis连接参数，定义派发频道前缀 CHANNEL_DISPATCH 与结果频道 CHANNEL_RESULT。Master会随机挑选编号为1~3的计算节点频道发送作业ID，并在后台持续监听结果频道打印完成信息。
代码实现（已修正原始缩进以确保语法正确）：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
import threading
import random
import redis

REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_DB = 0
CHANNEL_DISPATCH_PREFIX = 'CHANNEL_DISPATCH'
CHANNEL_RESULT = 'CHANNEL_RESULT'

class MyServerDispatchThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
        for i in range(1, 11):
            channel = f"{CHANNEL_DISPATCH_PREFIX}_{random.randint(1, 3)}"
            print(f"Dispatch job {i} to {channel}")
            ret = r.publish(channel, str(i))
            if ret == 0:
                print(f"Dispatch job {i} failed.")
            time.sleep(1)

class MyServerResultHandleThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
        p = r.pubsub()
        p.subscribe(CHANNEL_RESULT)
        for message in p.listen():
            if message['type'] != 'message':
                continue
            print(f"Received finished job {message['data']}")

class MyMaster():
    def start(self):
        MyServerResultHandleThread().start()
        MyServerDispatchThread().start()

if __name__ == "__main__":
    MyMaster().start()
    time.sleep(10000)

第2步：编写工作节点（Slave）代码

目的：让计算节点按需监听特定分发频道，执行任务后回写结果。入口为 MySlave.start()，默认创建3个Worker线程分别监听 CHANNEL_DISPATCH_1 到 _3。
动作参数：每个线程独立维护PubSub连接，收到消息后模拟2秒计算耗时，随即通过 r.publish(CHANNEL_RESULT, ...) 将作业号发回结果频道。
代码实现：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
import threading
import redis

REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_DB = 0
CHANNEL_DISPATCH_PREFIX = 'CHANNEL_DISPATCH'
CHANNEL_RESULT = 'CHANNEL_RESULT'

class MyJobWorkerThread(threading.Thread):
    def __init__(self, channel):
        threading.Thread.__init__(self)
        self.channel = channel

    def run(self):
        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
        p = r.pubsub()
        p.subscribe(self.channel)
        print(f"{self.channel}: Worker started.")
        for message in p.listen():
            if message['type'] != 'message':
                continue
            job_id = message['data']
            print(f"{self.channel}: Received dispatched job {job_id}")
            print(f"{self.channel}: Running dispatched job {job_id}")
            time.sleep(2)
            print(f"{self.channel}: Send finished job {job_id}")
            ret = r.publish(CHANNEL_RESULT, job_id)
            if ret == 0:
                print(f"{self.channel}: Result publish failed.")

class MySlave():
    def start(self):
        for i in range(1, 4):
            MyJobWorkerThread(f"{CHANNEL_DISPATCH_PREFIX}_{i}").start()

if __name__ == "__main__":
    MySlave().start()
    time.sleep(10000)

第3步：测试运行与完成判断

操作顺序如下：
1. 打开终端，运行Slave程序：
python slave.py
Slave启动后会创建3个工作线程并阻塞在 p.listen() 等待分发频道推送消息。
2. 另开一个终端，运行Master程序：
python master.py
Master会每隔1秒向随机选中的频道发布作业ID。Slave接收到后会打印执行日志，并将结果回传到 CHANNEL_RESULT。Master的结果处理线程捕获该消息后，会在控制台输出“Received finished job...”。

完成状态判断：当Master控制台稳定交替显示派发消息与结果接收记录，且无频繁断连或阻塞报错时，即表示调度通道已打通。注意，由于使用了纯内存Pub/Sub，若节点重启或连接断开，未消费的消息将丢失，生产环境需结合Redis Stream或持久化队列补充可靠性保障。

简要总结

本文通过Python标准库多线程与Redis原生发布订阅接口，快速搭建了一套可运行的轻量级作业调度原型。核心在于正确区分分发频道（用于任务投递）与结果频道（用于状态回传），并利用 publish() 的返回值辅助监控投递状态。以上就是利用Python和Redis Pub/Sub实现轻量级作业调度系统的详细内容，更多关于Python高级应用与分布式消息通信的资料请关注本站其它相关文章！