Как мне справиться с этим вариантом использования с помощью EventMachine?

Используйте один из драйверов Postgresql EM или EM.defer, чтобы не блокировать реактор.

Когда вы получаете сообщение «reload_credentials», просто переверните флаг, который приведет к тому, что все последующие сообщения будут поставлены в очередь. После завершения 'reload_credentials' обработайте все сообщения из очереди. После того, как очередь пуста, установите флаг, который заставляет сообщения обрабатываться по мере их получения.

Драйверы EM для Postgresql перечислены здесь: https://github.com/eventmachine/eventmachine/wiki/Protocol-Implementations

module Server
  def post_init
    @queue               = []
    @loading_credentials = false
  end

  def recieve_message(type, data)
    return @queue << [type, data] if @loading_credentials || [email protected]?
    return process_msg(type, data) unless :reload_credentials == type
    @loading_credentials = true
    reload_credentials do
      @loading_credentials = false
      process_queue
    end
  end

  def reload_credentials(&when_done)
    EM.defer( proc { query_and_load_credentials }, when_done )
  end


  def process_queue
    while (type, data = @queue.shift)
      process_msg(type, data)
    end
  end

  # lots of other methods
end

EM.start_server(HOST, PORT, Server)

Если вы хотите, чтобы все соединения помещались в очередь сообщений всякий раз, когда какое-либо соединение получает сообщение «reload_connections», вам придется координировать свои действия через собственный класс.

Я полагаю, что что-то вроде вашей текущей реализации:

    class Worker
      def initialize queue
        @queue = queue
        dequeue
      end

      def dequeue
        @queue.pop do |item|
          begin
            work_on item
          ensure
            dequeue
          end
        end
      end

      def work_on item
        case item.type
        when :reload_credentials
          # magic happens here
        else
          # more magic happens here
        end
      end
    end


    q = EM::Queue.new

    workers = Array.new(10) { Worker.new q }

Проблема выше, если я вас правильно понимаю, заключается в том, что вы не хотите, чтобы работники работали над новыми заданиями (заданиями, которые поступили раньше на временной шкале производителя), чем над любыми заданиями reload_credentials. Следующее должно обслуживать это (дополнительные слова предостережения в конце).

    class Worker
      def initialize queue
        @queue = queue
        dequeue
      end

      def dequeue
        @queue.pop do |item|
          begin
            work_on item
          ensure
            dequeue
          end
        end
      end

      def work_on item
        case item.type
        when :reload_credentials
          # magic happens here
        else
          # more magic happens here
        end
      end
    end

    class LockingDispatcher
      def initialize channel, queue
        @channel = channel
        @queue = queue

        @backlog = []
        @channel.subscribe method(:dispatch_with_locking)

        @locked = false
      end

      def dispatch_with_locking item
        if locked?
          @backlog << item
        else
          # You probably want to move the specialization here out into a method or
          # block that's passed into the constructor, to make the lockingdispatcher
          # more of a generic processor
          case item.type
          when :reload_credentials
            lock
            deferrable = CredentialReloader.new(item).start
            deferrable.callback { unlock }
            deferrable.errback  { unlock }
          else
            dispatch_without_locking item
          end
        end
      end

      def dispatch_without_locking item
        @queue << item
      end

      def locked?
        @locked
      end

      def lock
        @locked = true
      end

      def unlock
        @locked = false
        bl = @backlog.dup
        @backlog.clear
        bl.each { |item| dispatch_with_locking item }
      end

    end

    channel = EM::Channel.new
    queue = EM::Queue.new

    dispatcher = LockingDispatcher.new channel, queue

    workers = Array.new(10) { Worker.new queue }

Таким образом, входные данные для первой системы поступают на q, а в этой новой системе — на channel. queue по-прежнему используется для распределения работы между работниками, но queue не заполняется, пока выполняется операция обновления учетных данных. К сожалению, поскольку я не тратил больше времени, я не обобщил LockingDispatcher таким образом, чтобы он не был связан с типом элемента и кодом для отправки CredentialsReloader. Я оставлю это вам.

Вы должны отметить здесь, что, хотя это и служит тому, что я понимаю в отношении вашего первоначального запроса, как правило, лучше ослабить требования такого рода. Есть несколько нерешенных проблем, которые по существу не могут быть устранены без изменения этого требования:

• Система не ожидает завершения выполнения заданий перед запуском заданий учетных данных.
• Система будет очень плохо обрабатывать пакеты учетных данных — другие элементы, которые могут быть обработаны, не будут обрабатываться.
• В случае ошибки в коде учетных данных невыполненная работа может заполнить оперативную память и привести к сбою. Простого тайм-аута может быть достаточно, чтобы избежать катастрофических последствий, если код можно прервать, а последующие сообщения достаточно обрабатываются, чтобы избежать дальнейших взаимоблокировок.

На самом деле похоже, что у вас есть какое-то представление об идентификаторе пользователя в системе. Если вы продумаете свои требования, вполне возможно, что вам нужно будет занести в журнал только те элементы, которые относятся к идентификатору пользователя, учетные данные которого находятся в состоянии обновления. Это другая проблема, связанная с другим видом диспетчеризации. Попробуйте хэш заблокированных невыполненных работ для этих пользователей с обратным вызовом при завершении учетных данных, чтобы слить эти невыполненные работы в рабочие процессы, или что-то подобное.

Удачи!