如何让Ruby on Rails监听与其他应用共享的PostgreSQL表的新增记录？

针对你的需求（Rails 6.1 + PostgreSQL，跨应用插入记录后触发Rails业务逻辑，无法新增API端点），以下是两种健壮、可测试的方案，结合你的顾虑给出具体实现：

方案1：轮询标记字段（你提到的方案，优先推荐） #

适合对实时性要求不高的场景，是最稳妥的实现方式。

实现步骤 #

1. 给目标表新增processed布尔字段，默认未处理

   # 生成迁移文件
   rails generate migration AddProcessedToTargetTable processed:boolean
   # 编辑迁移文件，确保默认值与非空约束
   def change
     add_column :target_tables, :processed, :boolean, default: false, null: false
   end
   

2. 编写处理任务的Job类

   # app/jobs/process_new_records_job.rb
   class ProcessNewRecordsJob < ApplicationJob
     queue_as :default
   
     def perform
       TargetTable.where(processed: false).find_each do |record|
         record.with_lock do
           # 二次校验避免并发重复处理
           next if record.processed?
           # 执行你的业务逻辑
           handle_new_record(record)
           # 标记为已处理
           record.update!(processed: true)
         end
       end
     end
   
     private
   
     def handle_new_record(record)
       # 这里编写你的业务代码，比如数据同步、通知推送等
     end
   end
   

3. 用Whenever配置定时任务

   # config/schedule.rb
   every 5.minutes do # 根据业务需求调整间隔
     runner "ProcessNewRecordsJob.perform_now"
   end
   

   执行whenever --update-crontab生效定时任务

核心优势 #

• 健壮性：即使Rails应用临时宕机，重启后会自动处理所有未标记的记录，不会丢失任务
• 可测试性：直接针对ProcessNewRecordsJob写单元测试，模拟未处理记录即可验证逻辑
• 实现简单：无需依赖PostgreSQL高级特性，代码可读性高

方案2：优化版PostgreSQL LISTEN/NOTIFY（满足实时性需求） #

如果你需要实时触发逻辑，可以解决你担心的健壮性与可测试性问题，配合轮询兜底实现双保障。

实现步骤 #

1. 在PostgreSQL中创建触发器，插入记录时发送通知

   -- 创建通知函数
   CREATE OR REPLACE FUNCTION notify_new_target_record()
   RETURNS TRIGGER AS $$
   BEGIN
     PERFORM pg_notify('new_target_record_channel', NEW.id::text);
     RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;
   
   -- 给目标表绑定触发器
   CREATE TRIGGER trigger_new_target_record
   AFTER INSERT ON target_tables
   FOR EACH ROW EXECUTE FUNCTION notify_new_target_record();
   

2. 在Rails中编写常驻监听进程

   # lib/tasks/listen_new_records.rake
   namespace :listen do
     desc "Listen for new records in target_tables"
     task records: :environment do
       ActiveRecord::Base.connection_pool.with_connection do |conn|
         conn.execute("LISTEN new_target_record_channel")
         loop do
           conn.raw_connection.wait_for_notify do |_channel, _pid, payload|
             record = TargetTable.find_by(id: payload)
             if record && !record.processed?
               ProcessNewRecordsJob.perform_later(record)
             end
           end
         end
       end
     rescue ActiveRecord::ConnectionNotEstablished
       sleep 5
       retry
     end
   end
   

3. 用进程管理工具（如systemd、foreman）保障监听进程存活，避免意外退出

解决你的顾虑 #

• 健壮性：保留processed字段，监听进程宕机后，用方案1的轮询任务补处理未记录；进程守护工具确保进程自动重启
• 可测试性：业务逻辑依然通过ProcessNewRecordsJob测试，监听逻辑可在测试环境手动发送pg_notify验证触发流程

核心优势 #

• 实时性高：记录插入后立即触发业务逻辑
• 避免轮询的周期性数据库查询开销

总结建议 #

• 若对实时性要求不高，优先选择轮询方案，实现成本低且稳定性拉满
• 若需要实时处理，选择优化版LISTEN/NOTIFY+轮询兜底，兼顾实时性与健壮性

内容的提问来源于stack exchange，提问作者Paul