在作业处理期间改变 python-gearman 工作任务

Question

我正在尝试更改 python-gearman 工作人员在其工作周期中可用的任务。我这样做的原因是为了让我对我的工作进程有一点控制，并允许它们从数据库重新加载。我需要每个工作人员定期重新加载，但我不想简单地终止进程，并且我希望服务始终可用，这意味着我必须批量重新加载。因此，我将重新加载 4 个工作人员，同时另外 4 个工作人员可用于处理，然后重新加载接下来的 4 个工作人员。

过程：

1. 启动重新加载过程4次。
   1. 取消注册重新加载进程
   2. 重新加载数据集
   3. 注册一个finishReload任务
   4. 返回
2. 重复步骤 1，直到没有注册了 reload 任务的工作人员。
3. 启动 finishReload(1) 任务，直到没有可用的工作线程执行 finishReload 任务。

(1) finishReload任务注销finishReload任务并注册reload任务然后返回。

现在，我遇到的问题是，当我更改工作进程可用的任务时，作业会失败。 gearmand 日志中没有错误消息或异常，只有“ERROR”。这是一个复制该问题的快速程序。

工人

import gearman 
def reversify(gmWorker, gmJob): 
        return "".join(gmJob.data[::-1]) 
def strcount(gmWorker, gmJob): 
        gmWorker.unregister_task('reversify')  # problem line 
        return str(len(gmJob.data)) 
worker = gearman.GearmanWorker(['localhost:4730']) 
worker.register_task('reversify', reversify) 
worker.register_task('strcount', strcount) 
while True: 
        worker.work() 

客户

import gearman 
client = gearman.GearmanClient(['localhost:4730']) 
a = client.submit_job('reversify', 'spam and eggs') 
print a.result 
>>> sgge dna maps 

a = client.submit_job('strcount', 'spam and eggs') 
...

如果有任何我可以解释的事情，请告诉我。

编辑：我知道有人会要求查看我提到的日志。我也已将这个问题发布到 Google 上的 gearman 群组，并且日志可用那里。

Answer 1

乍一看，问题似乎在于您正在开始一项作业，然后在作业完成之前从作业服务器取消注册工作人员执行该作业的能力。

Answer 2

看起来子类化 GearmanWorker 类并添加一些标志可以解决这个问题。我需要先让作业完成，然后再开始从工作线程向服务器发出新命令，这似乎会中断当前作业。因此，如果我们覆盖 on_job_complete 函数，我们可以检查启用/禁用标志，并在发出 send_job_complete 命令后对这些标志采取行动。新的工人计划如下：

WORKER

import gearman

def reversify(gmWorker, gmJob):
        return "".join(gmJob.data[::-1])

def enable_reversify(gmWorker, gmJob):
        myWorker.enableReversify = 1
        return 'OK'

def strcount(gmWorker, gmJob):
        myWorker.enableReversify = -1
        return str(len(gmJob.data))

class myWorker(gearman.GearmanWorker):

        enableReversify = 0 # 0 = do nothing, -1 = turn off, 1 = turn on

        def on_job_complete(self, current_job, job_result):
                self.send_job_complete(current_job, job_result)
                ### check the flag here and enable or disable tasks ###
                if myWorker.enableReversify == -1:
                        self.unregister_task('reversify')
                if myWorker.enableReversify == 1:
                        self.register_task('reversify', reversify)
                myWorker.enableReversify = 0 # reset the flag
                return True

worker = myWorker(['localhost:4730']) 
worker.register_task('reversify', reversify)
worker.register_task('strcount', strcount)
worker.register_task('enableReversify', enable_reversify)

while True:
        worker.work()