# 定义发送邮件的 worker
class EmailWorker
  include Sidekiq::Worker
  def perform(email, content)
    Email.send(email, content) # 发送邮件
  end
end

# 调用端
class EmailController
  def welcome
    EmailWorker.perform_async("xxx@xx.com", "welcome to my home")
  end
end

为了排版以下内容只列出关键方法，均有删减
def perform_async(*args)
    # 1. 把类对象和参数组成 hash 后传递给 client_push
    client_push("class" => self, "args" => args)
end

def client_push(item)
    # 2. 实例化一个 client 把 item 参数传递给 push 方法
    Sidekiq::Client.new(pool).push(item)
end
def push(item)
  normed = normalize_item(item) # 格式化 hash
  payload = process_single(item["class"], normed) # 中间件方法，后面插曲有提到

  if payload
    raw_push([payload]) # 关键方法
    payload["jid"]
  end
end

def raw_push(payloads)
  @redis_pool.with do |conn|
    conn.multi do
      atomic_push(conn, payloads)
    end
  end
  true
end

# 最终处理的方法
def atomic_push(conn, payloads)
  # 是否定时任务，如果是定时任务，则按照定时任务的逻辑来处理字符串后放入队列
  if payloads.first["at"]
    conn.zadd("schedule", payloads.map { |hash|
      at = hash.delete("at").to_s
      [at, Sidekiq.dump_json(hash)]
    })
  else
    # 非定时任务，记录入队时间，把它放入 redis 的列表
    queue = payloads.first["queue"]
    now = Time.now.to_f
    to_push = payloads.map { |entry|
      entry["enqueued_at"] = now
      Sidekiq.dump_json(entry)
    }
    conn.sadd("queues", queue)
    conn.lpush("queue:#{queue}", to_push)
  end
end

完。

以下内容有删减
# 首先，打开了管道
self_read, self_write = IO.pipe 
# 注册信号量
sigs.each do |sig|
    trap sig do
      self_write.write("#{sig}\n")
    end
end

launcher.run # 主要方法

# 死循环一直读取管道，直到发送了信号量才进行 handle_signal，可参考后面的信号量说明
while (readable_io = IO.select([self_read]))
  signal = readable_io.first[0].gets.strip
  handle_signal(signal)
end

class Enq
  def enqueue_jobs(now = Time.now.to_f.to_s, sorted_sets = SETS)
    # A job's "score" in Redis is the time at which it should be processed.
    # Just check Redis for the set of jobs with a timestamp before now.
    Sidekiq.redis do |conn|
      sorted_sets.each do |sorted_set|
        # Get the next item in the queue if it's score (time to execute) is <= now.
        # We need to go through the list one at a time to reduce the risk of something
        # going wrong between the time jobs are popped from the scheduled queue and when
        # they are pushed onto a work queue and losing the jobs.
        while (job = conn.zrangebyscore(sorted_set, "-inf", now, limit: [0, 1]).first)

          # Pop item off the queue and add it to the work queue. If the job can't be popped from
          # the queue, it's because another process already popped it so we can move on to the
          # next one.
          if conn.zrem(sorted_set, job)
            Sidekiq::Client.push(Sidekiq.load_json(job))
            Sidekiq.logger.debug { "enqueued #{sorted_set}: #{job}" }
          end
        end
      end
    end
  end
end

def initialize(options = {})
  logger.debug { options.inspect }
  @options = options
  @count = options[:concurrency] || 10
  @done = false
  @workers = Set.new
  # 根据 count 的设置，来决定实例化多少个 `Processor` 对象，默认是 10 个
  @count.times do
    @workers << Processor.new(self)
  end
  @plock = Mutex.new
end

# Processor.new(self) 执行 options[:concurrency] 次 start
def start
  @workers.each do |x|
    x.start
  end
end

# 继续进入 Processor 类的 start 方法看

def start
  @thread ||= safe_thread("processor", &method(:run))
end

def run
  process_one until @done
  @mgr.processor_stopped(self)
rescue Sidekiq::Shutdown
  @mgr.processor_stopped(self)
rescue Exception => ex
  @mgr.processor_died(self, ex)
end

require 'sidekiq'

Sidekiq.configure_client do |config|
  config.redis = { db: 1 }
end

Sidekiq.configure_server do |config|
  config.redis = { db: 1 }
end

Sidekiq.configure_client do |config|
  class MyClientHook
    def call(worker_class, msg, queue, redis_pool)
      puts "Before push"
      result = yield
      puts "After push"
      puts result
      result
    end
  end
  config.client_middleware do |chain|
    chain.add MyClientHook
  end
end

class OurWorker
  include Sidekiq::Worker

  def perform(msg)
    puts "hello #{msg}"
  end
end
# 启动服务端
# bundle exec sidekiq -r ./woker.rb
# 启动客户端
# bundle exec irb -r ./woker.rb
# OurWorker.perform_async("abc") # 生产一条数据
# 输出： {"class"=>"OurWorker", "args"=>["abc"], "retry"=>true, "queue"=>"default", "jid"=>"966a57511c1d2b8d314b1318", "created_at"=>1563630849.305185}

# 示例代码01
module Skill
  def self.included(base)
    base.extend(ClassMethod)
    base.class_eval do
      def fly
        puts 'I can fly'
      end
    end
  end
  
  module ClassMethod
    def run
      puts  'Everybody can run'
    end
  end
end

class User
  include Skill
end

user = User.new
user.fly # 输出 I can fly

# 示例代码02
require 'active_support/concern'
module Skill
  extend ActiveSupport::Concern

  included do
    def fly
      puts 'I can fly'
    end
  end
  
  class_methods do
    def run
      puts  'Everybody can run'
    end
  end
end

class User
  include Skill
end

user = User.new
user.fly # 输出 I can fly

# 源代码01
# Define class methods from given block.
# You can define private class methods as well.
# ....
def class_methods(&class_methods_module_definition)
  mod = const_defined?(:ClassMethods, false) ?
    const_get(:ClassMethods) :
    const_set(:ClassMethods, Module.new)

  mod.module_eval(&class_methods_module_definition)
end

# 源码02
# Evaluate given block in context of base class,
# so that you can write class macros here.
# When you define more than one +included+ block, it raises an exception.
def included(base = nil, &block)
  if base.nil?
    if instance_variable_defined?(:@_included_block)
      if @_included_block.source_location != block.source_location
        raise MultipleIncludedBlocks
      end
    else
      @_included_block = block
    end
  else
    super
  end
end

module A
  def self.included(target)
    v = target.instance_methods.include?(:method_name)
    puts "in included: #{v}"
  end

  def self.append_features(target)
    v = target.instance_methods.include?(:method_name)
    puts "in append features before: #{v}"
    super
    v = target.instance_methods.include?(:method_name)
    puts "in append features after: #{v}"
  end

  def method_name
  end
end

class X
 include A
end

# 以上代码的输出为
# in append features before: false
# in append features after: true
# in included: true

def append_features(base) #:nodoc:
  if base.instance_variable_defined?(:@_dependencies)
    base.instance_variable_get(:@_dependencies) << self
    false
  else
    return false if base < self
    @_dependencies.each { |dep| base.include(dep) }
    super
    base.extend const_get(:ClassMethods) if const_defined?(:ClassMethods)
    base.class_eval(&@_included_block) if instance_variable_defined?(:@_included_block)
  end
end

module Foo
  def self.included(base)
    base.class_eval do
      def self.method_injected_by_foo
        ...
      end
    end
  end
end

module Bar
  def self.included(base)
    base.method_injected_by_foo
  end
end

class Host
  # 我实际上只想引用 Bar 模块
  # 引入 Foo 模块是因为 Bar 模块的某方法依赖于 Foo，所以我不得不在 Bar 之前引用一下 Foo
  include Foo 
  include Bar
end

module Foo
  def self.included(base)
    base.class_eval do
      def self.method_injected_by_foo
        ...
      end
    end
  end
end

module Bar
  include Foo
  def self.included(base)
    base.method_injected_by_foo
  end
end

class Host
  include Bar
end

undefined method `method_injected_by_foo' for Host:Class (NoMethodError)

require 'active_support/concern'

module Foo
  extend ActiveSupport::Concern
  included do
    def self.method_injected_by_foo
      ...
    end
  end
end

module Bar
  extend ActiveSupport::Concern
  include Foo

  included do
    self.method_injected_by_foo
  end
end

class Host
  include Bar # It works, now Bar takes care of its dependencies
end