ruby-on-rails - postgres LISTEN/NOTIFY rails

Question

Ryan Bates 在 this episode 中讨论推送通知时提到了 Postgres 的 LISTEN/NOTIFY 功能，但我还没有找到任何关于如何在我的 Rails 应用程序中实现 LISTEN/NOTIFY 的提示。

这是 wait_for_notify 的文档pg 适配器内部的功能，但我无法弄清楚它到底是做什么的/是为什么而设计的。

我们是否需要直接接入 pg 适配器的 connection 变量？

Answer 1

您正在使用 wait_for_notify 方法查找正确的位置，但由于 ActiveRecord 显然不提供使用它的 API，因此您需要获取底层的 PG::Connection ActiveRecord 用来与 Postgres 对话的对象(或者其中之一，如果你正在运行多线程设置)。

建立连接后，只需执行所需的任何LISTEN 语句，然后将 block (和可选的超时期限)传递给wait_for_notify。请注意，这将阻塞当前线程，并独占 Postgres 连接，直到达到超时或发生 NOTIFY(例如，您不希望在 Web 请求中执行此操作)。当另一个进程在您正在收听的 channel 之一上发出 NOTIFY 时，将使用三个参数调用该 block - 通知的 channel ，触发 的 Postgres 后端的 pid >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

有关更一般用法的示例，请参阅 Sequel's implementation .

编辑添加:这是对正在发生的事情的另一种描述。这可能不是幕后的确切实现，但它似乎足够好地描述了行为。

Postgres 为每个连接保留一个通知列表。当您使用一个连接来执行LISTEN channel_name 时，您是在告诉 Postgres 该 channel 上的任何通知都应该被推送到该连接的列表中(多个连接可以监听同一个 channel ，因此单个通知可以最终被推到许多列表中)。一个连接可以同时LISTEN到多个 channel ，并且对其中任何一个的通知都将被推送到同一个列表。

wait_for_notify 所做的是从连接列表中弹出最旧的通知并将其信息传递给 block - 或者，如果列表为空，则休眠直到通知可用，并且对那(或者直到达到超时，在这种情况下它只返回零)。由于 wait_for_notify 只处理一个通知，如果你想处理多个通知，你将不得不重复调用它。

当您UNLISTEN channel_name 或UNLISTEN * 时，Postgres 将停止将这些通知推送到您的连接列表，但已经推送到该列表的通知将保留在那里, 而 wait_for_notify 在下次调用时仍会返回它们。这可能会导致一个问题，即在 wait_for_notify 之后但在 UNLISTEN 之前累积的通知仍然存在，并且在另一个线程检查该连接时仍然存在。在这种情况下，在 UNLISTEN 之后，您可能希望调用 wait_for_notify 并在短时间内超时，直到它返回 nil。但是，除非您出于许多不同的目的大量使用 LISTEN 和 NOTIFY，否则可能不值得担心。

我在上面添加了一个更好的链接到 Sequel 的实现，我建议查看它。这非常简单。