利用 ActiveSupport::Notifications 在 Rails 中实现 PUB/SUB 模式

前言

自从 Rails 4 从核心中移除了 Observers 之后，对于复杂的业务逻辑和依赖关系该放在哪，大家就开始各显神通了。

有人建议利用 Concerns 和 callback 来搞定。

module MyConcernModule
  extend ActiveSupport::Concern

  included do
    after_save :do_something
  end

  def do_something
     ...
  end
end

class MyModel < ActiveRecord::Base
  include MyConcernModule
end

更多的人认为，依据单一职责原则抽出一个 Service Object 才是王道。

在此基础上，Wisper确实是一个很好的解决方案。

你可以这么玩：

class PostsController < ApplicationController
  def create
    @post = Post.new(params[:post])

    @post.subscribe(PusherListener.new)
    @post.subscribe(ActivityListener.new)
    @post.subscribe(StatisticsListener.new)

    @post.on(:create_post_successful) { |post| redirect_to post }
    @post.on(:create_post_failed)     { |post| render :action => :new }

    @post.create
  end
end

也可以这么玩：

class ApplicationController < ActionController::Base
  around_filter :register_event_listeners

  def register_event_listeners(&around_listener_block)
    Wisper.with_listeners(UserListener.new) do
      around_listener_block.call
    end
  end
end

class User
  include Wisper::Publisher
  after_create{ |user| publish(:user_registered, user) }
end

class UserListener
  def user_registered(user)
    Analytics.track("user:registered", user.analytics)
  end
end

但是 @Rei 一句 Rails 有 ActiveSupport::Notifications 竟让我无言以对。

ActiveSupport::Notifications

在阅读了相关阅读中给出的链接，以及 Notifications in Rails 3 和官方文档之后，感觉这东西设计出来完全是为了进行统计、日志、性能分析之类的事情啊。

那么用它能不能实现一个 PUB/SUB 模式呢？先来回顾一下 PUB/SUB 模式核心的两个要点。

1. Publishers 在对象状态改变且需要触发事件的时候发布事件。
2. Subscribers 仅接收它们能响应的事件，并且在每个事件中可以接收到被监控的对象。

The Basics of AS::Notifications

ActiveSupport::Notifications 主要核心就是两个方法：instrument 和 subscribe。

你可以把 instrument 理解为发布事件。instrument 会在代码块执行完毕并返回结果之后，发布事件 my.custom.event，同时会自动把相关的一组参数：开始时间、结束时间、每个事件的唯一 ID 等，放入 payload 对象。

ActiveSupport::Notifications.instrument "my.custom.event", this: :data do
  # do your custom stuff here
end

现在你可以监听这个事件：

ActiveSupport::Notifications.subscribe "my.custom.event" do |name, started, finished, unique_id, data|
  puts data.inspect # {:this=>:data}
end

理解了这两个方法，我们可以试着实现一个 PUB/SUB 模式。

Publisher

# app/pub_sub/publisher.rb
module Publisher
  extend self

  # delegate to ActiveSupport::Notifications.instrument
  def broadcast_event(event_name, payload={})
    if block_given?
      ActiveSupport::Notifications.instrument(event_name, payload) do
        yield
      end
    else
      ActiveSupport::Notifications.instrument(event_name, payload)
    end
  end
end

首先我们定了 Publisher。它把 payload 夹在事件中广播出去，代码块也可以当做一个可选参数传递进来。

我们可以在 model 或 controller 中发布具体事件。

if user.save
  # publish event 'user.created', with payload {user: user}
  Publisher.broadcast_event('user.created', user: user)
end

def create_user(params)
  user = User.new(params)
  # publish event 'user.created', with payload {user: user}, using block syntax
  # now the event will have additional data about duration and exceptions
  Publisher.broadcast_event('user.created', user: user) do
    User.save!
    # do some more important stuff here
  end
end

Subscriber

Subscriber 可以订阅事件，并将代码块当做参数，传递给 ActiveSupport::Notifications::Event 的实例。

# app/pub_sub/subscriber.rb
module Subscriber
  # delegate to ActiveSupport::Notifications.subscribe
  def self.subscribe(event_name)
    if block_given?
      ActiveSupport::Notifications.subscribe(event_name) do |*args|
        event = ActiveSupport::Notifications::Event.new(*args)
        yield(event)
      end
    end
  end
end

# subscriber example usage
Subscriber.subscribe('user.created') do |event|
  error = "Error: #{event.payload[:exception].first}" if event.payload[:exception]
  puts "#{event.transaction_id} | #{event.name} | #{event.time} | #{event.duration} | #{event.payload[:user].id} | #{error}"
end

经典场景

用户注册后，为该用户发送欢迎邮件。

# app/pub_sub/publisher.rb
module Publisher
  extend ::ActiveSupport::Concern
  extend self

  included do
    # add support for namespace, one class - one namespace
    class_attribute :pub_sub_namespace

    self.pub_sub_namespace = nil
  end

  # delegate to class method
  def broadcast_event(event_name, payload={})
    if block_given?
      self.class.broadcast_event(event_name, payload) do
        yield
      end
    else
      self.class.broadcast_event(event_name, payload)
    end
  end

  module ClassMethods
    # delegate to ASN
    def broadcast_event(event_name, payload={})
      event_name = [pub_sub_namespace, event_name].compact.join('.')
      if block_given?
        ActiveSupport::Notifications.instrument(event_name, payload) do
          yield
        end
      else
        ActiveSupport::Notifications.instrument(event_name, payload)
      end
    end
  end
end

# app/pub_sub/publishers/registration.rb
module Publishers
  class Registration
    include Publisher

    self.pub_sub_namespace = 'registration'
  end
end

# broadcast event
if user.save
  Publishers::Registration.broadcast_event('user_signed_up', user: user)
end

# app/pub_sub/subscribers/base.rb
module Subscribers
  class Base
    class_attribute :subscriptions_enabled
    attr_reader :namespace

    def initialize(namespace)
      @namespace = namespace
    end

    # attach public methods of subscriber with events in the namespace
    def self.attach_to(namespace)
      log_subscriber = new(namespace)
      log_subscriber.public_methods(false).each do |event|
        ActiveSupport::Notifications.subscribe("#{namespace}.#{event}", log_subscriber)
      end
    end

    # trigger methods when an even is captured
    def call(message, *args)
      method  = message.gsub("#{namespace}.", '')
      handler = self.class.new(namespace)
      handler.send(method, ActiveSupport::Notifications::Event.new(message, *args))
    end
  end
end

# app/pub_sub/subscribers/registration_mailer.rb
module Subscribers
  class RegistrationMailer < ::Subscribers::Base
    def user_signed_up(event)
      # lets delay the delivery using delayed_job
      RegistrationMailer.delay(priority: 1).welcome_email(event.payload[:user])
    end
  end
end

# config/initializers/subscribers.rb
Subscribers::RegistrationMailer.attach_to('registration')

醉了吗？详细的解释可以看相关阅读中的文章。如果你的队友没有把 ASN 吃透，肯定会掀桌子。

所以是引入一个 gem 还是自己根据 ASN 来实现同样的功能，还是由你自己想吧。

相关阅读

• Digging Deep with ActiveSupport::Notifications 视频
• Digging Deep with ActiveSupport::Notifications PDF
• Active Support Instrumentation
• Implementing PUB/SUB in Rails; using ActiveSupport::Notifications
• Instrumenting Your Code With ActiveSupport Notifications
• Using ActiveSupport::Notifications and ActiveSupport::Concern To Create An Audit Trail

4.22 补充

22 楼的 @satzcoal 提了如下几个问题，我也答应 @billy 在踩坑之后过来补充一下此文。@satzcoal 面对的问题有下面这些：

ActiveSupport::Notifications 在 sub 的管理上非常困难

ASN 对于 sub 的管理和 Wisper 其实并无太大差异。仍然可以进行全局订阅和临时订阅。如果你觉得难以管理，那我建议你跟我一样，在 initializes 文件夹下面建立一个 subscribers.rb 用来统一进行订阅。

subsribers = {
  todos: [ 'started', 'paused', 'completed', 'deleted' ],
  topics: ['created', 'deleted', 'sticked', 'unsticked', 'commented'],
  documents: ['created', 'deleted', 'updated', 'commented', 'recovered'],
}

subsribers.each do |key, value|
  value.each do |action|
    ActiveSupport::Notifications.subscribe("#{key}.#{action}") do |*args|
      event = ActiveSupport::Notifications::Event.new(*args)
      WebhookService.call(event.name, event.transaction_id, event.payload)
    end
  end
end

很难验证 sub 的重复性

如果你使用了上面那种集中式管理 sub 的方法，这个问题就略过吧。

很难进行 unsub

可以利用 unsubscribe 来取消订阅，但是我在使用中，没有遇到这个场景。

很难管理 sub 的顺序

我猜你问得是很难定义 instrument 的顺序吧。从别人那里摘抄一个代码，你看看是不是你需要的？

ActiveSupport::Notifications.instrument 'user.signup' do
  # create user record
  ActiveSupport::Notifications.instrument 'twitter.location' do
    user.last_location = Twitter.user(user.twitter_username).location
  end
  # do more work
end

很难对 sub 中的异常进行处理

同上直接上代码了

ActiveSupport::Notifications.instrument 'twitter.location', :user => user.id do
  user.last_location = Twitter.user(user.twitter_username).location
end

# instrument message data - name, start, end, id, payload
=> ['twitter.location', 2015-04-22 17:15:44, 2015-04-22 17:15:44, 'LAiKEjiMCy8XYY9Y'， { :user => 1023, :exception => ["Twitter::Error:NotFound", "not found"]}]

ASN 会在 payload 中返回 Exception Type 和 Exception Message 的。

小结

向 @billy 的报告。过去几周我使用 ASN 为我司的项目添加了 Webhook 功能。还是原来的答案，核心逻辑不该用 ASN。

ASN 可以使用在下列场景：

• Custom Logging
• New Relic Tracking
• Google Analytics events
• Gampfire / HipChat alerts
• Schedule background jobs
• Fire web hooks