使用Sidekiq在另一个worker中调用worker

Question

我在使用Sidekiq时遇到了调用另一个worker的问题。 我的两个worker代码如下：

class UserExportWorker
  include Sidekiq::Worker

  sidekiq_options queue: :users, unique: true, retry: false

  def perform(user_id)
    # code that exports user to external service via API
    UserExportStatusWorker.perform_in(1.minute, user_export_id, user_id)
  end
end

class UserExportStatusWorker
  include Sidekiq::Worker

  sidekiq_options queue: :users, unique: true, retry: false

  def perform(user_export_id, user_id)
    # code that check if exporting user to external service via API is finished
    if export.completed?
      user.update(status: 'exported')
    else
      UserExportStatusWorker.perform_in(1.minute, user_export_id, user_id)
    end
  end
end

UserExportWorker通过API将用户导出到某些外部服务，并运行UserExportStatusWorker，该工作程序将检查外部API中的导出是否完成，如果没有完成，则会再次启动相同的工作程序，当导出完成后，它将更新数据库中的用户状态列。问题是，即使成功将用户导出到外部API，UserExportStatusWorker也不会在数据库中更新用户。但当我将代码更改为以下内容时：

UserExportWorker将用户导出到某些外部服务，然后直接更新数据库中的用户状态列。这样做的话，就不需要使用UserExportStatusWorker了。

class UserExportWorker
  include Sidekiq::Worker

  sidekiq_options queue: :users, unique: true, retry: false

  def perform(user_id)
    # code that exports user to external service via API
    UserExportStatusWorker.new.perform(user_export_id, user_id)
  end
end

class UserExportStatusWorker
  include Sidekiq::Worker

  sidekiq_options queue: :users, unique: true, retry: false

  def perform(user_export_id, user_id)
    # code that check if exporting user to external service via API is finished
    if export.completed?
      user.update(status: 'exported')
    else
      UserExportStatusWorker.new.perform(user_export_id, user_id)
    end
  end
end

我完全没有头绪…为什么这个不起作用呢？

- Mateusz Urbański

---

你确定UserExportStatusWorker中的任务正在使用perform_in被加入到队列中吗？ - Anthony

在 Sidekiq 中，我有一个可视化界面，可以提供有关当前作业的信息。当我使用 perform_in 时，在 Sidekiq 管理面板中可以看到作业将在1分钟内计划执行。但是之后什么也没有发生。作业从计划作业中消失，但成功完成的作业计数器没有改变。 - Mateusz Urbański

如果您没有使用错误处理服务，我建议您暂时将 retry: true 设置为真，并观察UI中的重试队列以查看作业是否失败。我使用Rollbar，但他推荐了一些其他的 - https://github.com/mperham/sidekiq/wiki/Error-Handling - Anthony

@Anthony，当我在UserExportStatusWorker中将unique更改为false时，一切都按预期工作。 - Mateusz Urbański

独特的功能来自哪里？又是一个宝石吗？ - Anthony

Answer 1

理想情况下，如果要重新启动工作进程，您应该引发一个错误，并让 Sidekiq 进行重试，具体请参见docs。

所以，在您的情况下，在 UserExportWorker 类中，您可以编写一个方法来检查导出是否成功运行。如果没有成功，就抛出自定义的 ExportNotSuccessfulError 错误。

- Bruno Peres