Postgres LISTEN/NOTIFY рельсы

Ответ 1

Вы смотрите в нужном месте с помощью метода wait_for_notify, но поскольку ActiveRecord, по-видимому, не предоставляет API для его использования, вам нужно будет получить базовый объект PG:: Connection (или один из их, если вы используете многопоточную настройку), которые ActiveRecord использует для связи с Postgres.

После того, как вы получили соединение, просто выполните любые операторы LISTEN, которые вам нужны, затем передайте блок (и дополнительный период ожидания) в wait_for_notify. Обратите внимание, что это заблокирует текущий поток и монополизирует соединение Postgres, пока не будет достигнут тайм-аут или не произойдет NOTIFY (поэтому вы не захотите делать это в веб-запросе, например). Когда другой процесс выдает NOTIFY на одном из каналов, которые вы слушаете, блок будет вызываться с тремя аргументами - каналом, который был уведомлен, pid бэкэнд Postgres, который вызвал NOTIFY, и полезную нагрузку которые сопровождали NOTIFY (если они есть).

Я не использовал ActiveRecord довольно долго, так что может быть более чистый способ сделать это, но, похоже, это работает в 4.0.0.beta1:

# Be sure to check out a connection, so we stay thread-safe.
ActiveRecord::Base.connection_pool.with_connection do |connection|
  # connection is the ActiveRecord::ConnectionAdapters::PostgreSQLAdapter object
  conn = connection.instance_variable_get(:@connection)
  # conn is the underlying PG::Connection object, and exposes #wait_for_notify

  begin
    conn.async_exec "LISTEN channel1"
    conn.async_exec "LISTEN channel2"

    # This will block until a NOTIFY is issued on one of these two channels.
    conn.wait_for_notify do |channel, pid, payload|
      puts "Received a NOTIFY on channel #{channel}"
      puts "from PG backend #{pid}"
      puts "saying #{payload}"
    end

    # Note that you'll need to call wait_for_notify again if you want to pick
    # up further notifications. This time, bail out if we don't get a
    # notification within half a second.
    conn.wait_for_notify(0.5) do |channel, pid, payload|
      puts "Received a second NOTIFY on channel #{channel}"
      puts "from PG backend #{pid}"
      puts "saying #{payload}"
    end
  ensure
    # Don't want the connection to still be listening once we return
    # it to the pool - could result in weird behavior for the next
    # thread to check it out.
    conn.async_exec "UNLISTEN *"
  end
end

Пример более общего использования см. в Последовательная реализация.

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

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

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

Когда вы UNLISTEN channel_name или UNLISTEN *, Postgres перестанут нажимать эти уведомления на ваш список соединений, но те, которые уже были перенесены в этот список, останутся там, и wait_for_notify все равно вернет их, когда будет вызван следующий вызов, Это может привести к возникновению проблем, когда уведомления, которые накапливаются после wait_for_notify, но до UNLISTEN, остаются и остаются, когда другой поток проверяет это соединение. В этом случае после UNLISTEN вы можете вызвать wait_for_notify с короткими таймаутами, пока не вернет нуль. Но, если вы не используете большие возможности LISTEN и NOTIFY для разных целей, это, вероятно, не стоит беспокоиться.

Я добавил лучшую ссылку на реализацию Sequel выше, я бы рекомендовал посмотреть на нее. Это довольно просто.