如何在分布式环境中处理 PostgreSQL 触发器

Question

我们正在实施 PostgreSQL 触发器来监视多个表上的插入/更新/删除，以便另一个监听这些事件的应用程序可以使我们的关系数据库与我们的全文搜索数据库保持同步。

下面是触发器函数的样子：

CREATE FUNCTION notification() RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify('search', TG_TABLE_NAME || ',id,' || NEW.id);
  RETURN NULL;
END;
$$ LANGUAGE plpgsql;

这是我们向每个表添加触发器的方式：

CREATE TRIGGER foo_trigger AFTER INSERT OR UPDATE or DELETE ON foo
FOR EACH ROW EXECUTE PROCEDURE notification();

这是一个非常基本的示例，说明我们如何让节点应用程序（工作程序）侦听这些触发事件：

var pg  = require('pg');

var connString = "postgres://user@localhost/foo_local";

pg.connect(connString, function(err, client, done) {

  client.on('notification', function(msg) {
    //get the added / updated / deleted record
    //sync it with the search database
  });

  var query = client.query('LISTEN search');
});

这是我的三部分问题：

第 1 部分 我们的应用在多个实例之间进行负载平衡。当节点/工作应用程序（也是分布式的）接收到事件时会发生什么？正在侦听的工作应用程序的所有实例都会收到触发事件吗？

如果是这样，那就不好了——我们不希望工作应用程序的所有实例都处理每个事件，因为它们都在做相同的工作，这会抵消使用多个侦听器来分配负载的好处。我们如何缓解这种情况？

第 2 部分 如果 worker 收到触发事件，但它是长时间运行的，会发生什么？PostgreSQL 是否会将已触发的事件排队，直到侦听器收到它们？

第 3 部分 我们有大约 5 个表，希望在 INSERT/UPDATE/DELETE 上触发触发器。我们有很多请求，所以这会在短时间内触发很多事件。我们需要一个 worker 来监听这些事件并处理更改的记录，以便它可以将它们发送到全文搜索数据库。有没有更好的方法来构建它来处理卷？

我们团队正在考虑的另一个解决方案是放弃 SQL 触发器，只使用消息队列系统将消息推送到数据存储（SQS 或 Redis）中，然后让工作人员从队列中挑选消息。如果可以，我们希望避免这条路线，因为它为我们的平台添加了更多架构；但是，如果这是我们唯一的选择，我们准备这样做。

您的想法将不胜感激。

Answer 1

首先，在您的触发函数中，您可能希望通过提供有关更改内容的更具体的详细信息（例如在更新中）来让听众的生活更轻松。

你可以这样做：

CREATE OR REPLACE FUNCTION notification() RETURNS trigger AS $$
DECLARE
  id bigint;
BEGIN
  IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
    id = NEW.id;
  ELSE
    id = OLD.id;
  END IF;

  IF TG_OP = 'UPDATE' THEN
    PERFORM pg_notify('table_update', json_build_object('schema', TG_TABLE_SCHEMA, 'table', TG_TABLE_NAME, 'id', id, 'type', TG_OP, 'changes', hstore_to_json(hstore(NEW) - hstore(OLD)))::text);
    RETURN NEW;
  END IF;

  IF TG_OP = 'INSERT' THEN
    PERFORM pg_notify('table_update', json_build_object('schema', TG_TABLE_SCHEMA, 'table', TG_TABLE_NAME, 'id', id, 'type', TG_OP, 'row', row_to_json(NEW))::text);
    RETURN NEW;
  END IF;

  IF TG_OP = 'DELETE' THEN
    PERFORM pg_notify('table_update', json_build_object('schema', TG_TABLE_SCHEMA, 'table', TG_TABLE_NAME, 'id', id, 'type', TG_OP, 'row', row_to_json(OLD))::text);
    RETURN OLD;
  END IF;

END;
$$ LANGUAGE plpgsql;

现在回答您的问题，或者至少： 第 1 部分：我相信正在侦听的工作应用程序的所有实例都会收到触发的事件。这对于向多个侦听器发出发布/订阅风格的实时通知非常有用。对于您的用例，听起来您需要在基本的 PostgreSQL LISTEN/NOTIFY 之上添加某种队列包，例如queue_classic（对于 Ruby）或可能对于 node.js 的pg-jobs。

不管怎样，既然你问这个问题已经有几个月了，我想知道你最后走了哪条路，结果如何？您能分享一下您的经验和见解吗？