Python和Redis:Manager/Worker应用程序最佳实践

Question

关于使用Python和Redis创建用于运行异步命令的作业队列应用程序，我有一些一般性的问题。以下是我目前生成的代码：

def queueCmd(cmd):
    r_server.rpush("cmds", cmd)

def printCmdQueue():
    print r_server.lrange("cmds", 0 , -1)

def work():
    print "command being consumed: ", r_server.lpop("cmds")
    return -1

def boom(info):
    print "pop goes the weasel"

if __name__ == '__main__':

    r_server = redis.Redis("localhost")

    queueCmd("ls -la;sleep 10;ls")
    queueCmd("mkdir test; sleep 20")
    queueCmd("ls -la;sleep 10;ls")
    queueCmd("mkdir test; sleep 20")
    queueCmd("ls -la;sleep 10;ls")
    queueCmd("mkdir test; sleep 20")

    printCmdQueue()

    pool = Pool(processes=2)

    print "cnt:", +r_server.llen("cmds")
    #while r_server.llen("cmds") > 0:
    while True:
        pool.apply_async(work, callback=boom)
        if not r_server.lrange("cmds", 0, -1):
        #if r_server.llen("cmds") == 0:
            print "Terminate pool"
            pool.terminate()
            break

    printCmdQueue()

首先，我是否正确地认为，如果我需要与经理进行任何沟通，我希望通过回拨来进行沟通？我在这个例子中看到的快速示例将异步调用存储在一个结果中，并通过结果.get（超时=1）。通过交流，我的意思是把东西放回redis列表中。在

编辑：如果命令以异步方式运行，而我在main中的结果上超时，那么这是使工作进程超时还是只在管理器中超时？如果经理不能用这个来检查工人的出口代码就好了？在

接下来，此代码生成以下输出：

^{pr2}$

为什么工作人员希望一次使用多个cmd，即使我一次只弹出一个命令？在类似的not中，这并不总是很好地结束，有时需要使用ctrl+c。我认为这与apply泳sync（）和是否退出循环有关。我想知道工人方面是否需要更多的工作？在

如果我把ifs改成注释掉的那个，我得到：

ValueError: invalid literal for int() with base 10: 'ls -la;sleep 10;ls'

这似乎是检查是否需要中断的更好方法，但似乎函数有时返回字符串文本？在

如有任何改进建议，我们将不胜感激。我只是想做一个类似于linux机器上的服务/守护进程的管理器。它将用于从redis列表中获取作业（当前是命令，但可能更多），并将结果返回到redis列表中。然后，GUI将与这个管理器交互以获取队列的状态并返回结果。在

谢谢

编辑：

我意识到我有点傻。我不需要从worker访问redis服务器，这导致了一些错误（特别是ValueError）。在

要解决此问题，循环现在是：

while not r_server.llen("cmds") == 0:
    cmd = r_server.lpop("cmds")
    pool.apply_async(work, [cmd])

在这些行之后，我调用pool.close()。我用os.getpid()和os.getppid()来检查我是否确实有多个孩子在到处跑。在

如果这听起来是创建一个使用redis的manager/worker应用程序的好方法，我还是会很高兴的。在

Answer 1

您的问题是您试图在一个redis连接中同时运行多个命令。在

你期待的是

Thread 1     Thread 2
LLEN test    
1                            
LPOP test   
command      
             LLEN test
             0

但是你得到了

^{2}$

结果以相同的顺序返回，但没有将线程或命令链接到特定结果。单独的redis连接不是线程安全的-每个工作线程都需要一个。在

如果不恰当地使用流水线，您也会看到类似的问题—它是为只写的情况而设计的，例如向列表中添加许多项，在这种情况下，可以通过假设LPUSH成功而不是等待服务器在每个项之后通知您成功来提高性能。Redis仍将返回结果，但不一定是上一次发送命令的结果。在

除此之外，基本方法是合理的。不过，您可以做一些增强：

• 不要检查长度，只需使用非阻塞LPOP-如果它返回null，则列表为空
• 添加一个计时器，这样如果列表为空，它将等待，而不是仅仅发出另一个命令。在
• 在while循环条件中包含取消检查
• 处理连接错误-我使用一个外部循环设置，这样如果连接失败，worker将尝试重新连接（基本上是重新启动main），然后再完全终止工作进程。在