在Rails应用中使用GCP Pub/Sub流式订阅遇问题求助

1. 关于backup_poller.cc错误的解释 #

那个日志信息其实是gRPC库的非致命调试日志，不是导致程序异常退出的直接原因。它表示gRPC内部的定时器系统正在关闭，通常发生在你的worker进程启动listener后，没有保持运行状态就直接退出了——你的run_worker!方法里启动listener后，没有阻塞主线程的逻辑，进程启动listener后立刻结束，触发了gRPC的资源清理日志，所以你才会看到"Running worker"后立刻出现这条信息。

2. 在Rails中正确使用流式订阅（针对v0.29.0版本） #

你的google-cloud-pubsub版本是v0.29.0，这个版本的流式订阅API和后续新版本有差异，而且你提到的Google::Cloud::Pubsub::V1::Subscriber在该版本中还未引入（V1客户端是后续版本才添加的低阶API）。针对你的环境，修复并正确实现流式订阅的方法如下：

修复现有Worker代码 #

修改run_worker!方法，添加阻塞逻辑让进程保持运行，同时完善消息处理的错误捕获：

def run_worker!(queue_name = 'default')
  puts 'Running worker'
  topic = pubsub.topic(queue_name)
  # 确保订阅存在，不存在则自动创建
  subscription = topic.subscription("#{queue_name}_task") || topic.subscribe("#{queue_name}_task")
  
  listener = subscription.listen do |message|
    Rails.logger.info "Processing job: #{message.data}"
    begin
      # 解析Job类并执行任务
      job_class = message.data.constantize
      # 反序列化参数，避免类型丢失
      job_args = JSON.parse(message.attributes['arg'])
      job_class.perform(*job_args)
      message.acknowledge!
    rescue => e
      Rails.logger.error "Job processing failed: #{e.message}\n#{e.backtrace.join("\n")}"
      # 处理失败，让消息重新入队
      message.nacknowledge!
    end
  end

  listener.start
  # 添加信号捕获，支持优雅停止
  trap('INT') { listener.stop }
  trap('TERM') { listener.stop }
  # 阻塞主线程，保持Worker运行
  listener.wait_until_stopped
end

关键修改点说明 #

• 阻塞主线程：通过listener.wait_until_stopped让进程持续运行，不会启动listener后立刻退出；
• 信号处理：捕获INT和TERM信号，让Worker可以优雅停止，避免强制中断导致消息丢失；
• 参数序列化优化：确保Job参数通过JSON序列化/反序列化传递，避免复杂类型在Pub/Sub传输中丢失；
• 错误处理：添加异常捕获，失败时通过nacknowledge!让消息重新入队，提升任务可靠性。

Enqueue方法小优化 #

你的入队方法中参数传递存在类型丢失风险，建议将Job参数序列化为JSON字符串：

def enqueue(job)
  Rails.logger.info "[PubSubQueueAdapter enqueue job #{job.inspect}"
  topic = PubSubQueueAdapter.pubsub.topic(job.queue_name)
  # 将参数序列化为JSON，避免类型问题
  topic.publish(job.class.name, arg: JSON.dump(job.arguments))
end

3. 关于无法使用SubscriberClient的说明 #

在google-cloud-pubsub v0.29.0版本中，Google::Cloud::Pubsub::V1::SubscriberClient还未被引入，这个低阶API是在v1.x及以上版本才添加的。如果一定要使用流式Pull的低阶API，需要升级依赖版本，但考虑到你的Rails 5.1.4和Ruby 2.4.2版本，升级可能带来兼容性问题，建议先尝试修复现有代码来满足需求。

额外注意事项 #

• 确保GCP服务账号拥有Pub/Sub的发布者和订阅者权限；
• 验证project_id和auth_file配置正确，服务账号密钥文件路径可正常读取；
• 生产环境建议使用foreman等进程管理工具运行Worker，避免终端关闭后进程退出。

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