狠狠撸

Sidekiq 源码分析
zhangyuan
13年9月17日星期二

不谈什么
? 不谈 Celluloid
? 不谈 Sidekiq 的 web 界面
? 不谈 Sidekiq 的测试代码

阅读源码的好处

如何阅读源码
? 从功能（尤其是常用的用法）寻找线索
? 工具
? grep
? VIM、ctags、vim-scripts/taglist.vim

简单介绍
? 多线程的后台任务工具。
? 虽然 Sidekiq 是多线程的，但没有过多地
显式使用线程库，而是通过 Celluloid 实
现并发。Celluloid 是?一个实现了 Actor
Model 并发模型的 Ruby 库，它不仅仅隐
藏了线程的细节。
? 本幻灯片适用于 sidekiq 2.11.2 版本。

整体架构
? 客户端
? 任务入队
? 服务器端
? 读取任务并处理

任务分类
? 普通任务
? 放入后台后执行
? 定时任务
? 在某个时刻执行

客户端（1）
? 消息入队
? 搜索 “def perform_async” / “def perform_in”
? lib/sidekiq/worker.rb
? 序列化任务信息后，使用集合（set）保存队列
名称
? SADD key member [member ...]
? key 就是 queues

客户端（2）
? 普通队列使用不同的列表（list）保存任务
? LPUSH key value [value ...]
? key 是队列名称，以 “queue:” 为前缀
? 定时任务使用有序集合（SortedSet），保存在同?一个名称为 schedule 的
有序集合中
? ZADD key score member [[score member] [score member] ...]
? key 是队列名, score 是执行时刻，member 是任务信息

服务器
? 服务器端在命令行执行
? 接下来都是服务器端

解析命令行参数
? CLI - Command-Line Interface
? 可执行文件?一般在 bin/ 目录里
? 代码?一般会在?一个叫 cli.rb 的文件里
? 通常使用 optparse 来解析命令参数

解析队列和权重参数
:queues:
- [default, 2]
- [high, 5]

根据权重选取任务
queues_and_weights = [["default", 2], ["high", 5]]
queues = ["default", "default", "high", "high", "high", "high", "high"]
queues = ["queue:default", "queue:default", "queue:high", "queue:high", "queue:high", "queue:high",
"queue:high"]
queues_cmd = queues.shuf?e.uniq
queues_cmd << Sidekiq::Fetcher::TIMEOUT
Sidekiq.redis { |conn| conn.brpop(*queues_cmd) }

选取任务的代码
? lib/sidekiq/cli.rb
? Sidekiq::CLI#parse_con?g
? Sidekiq::BasicFetch#initialize
? Sidekiq::BasicFetch#retrieve_work

队列权重小结
? 将队列权重问题转换为概率问题，来选
取任务

处理系统信号
self_read, self_write = IO.pipe
%w(INT TERM USR1 USR2 TTIN).each do |sig|
trap sig do
self_write.puts(sig)
end
end
while readable_io = IO.select([self_read])
signal = readable_io.?rst[0].gets.strip
handle_signal(signal)
end

处理系统信号小结
? 使用 trap {} 注册系统信号
? 使用 IO.select 等待IO就绪，使用while循环，进
程不退出
? 只有?一个 self_read ，能否改成阻塞IO？
? 为什么要使用 IO.pipe 和 select ，而不是直接
使用 loop {}
? 参考书籍 Working With Unix Processes

修改程序名称
? $0 显示当前的worker状态
? $PROGRAM
? 定时刷新
? Sidekiq::Manager#procline

任务的分类处理
? 普通任务
? Sidekiq::Manager#async.start
? 使用列表（list）存取
? 定时任务
? Sidekiq::Scheduled::Poller#async.poll(true)
? 使用有序集合（SortedSet）存取

普通任务
? N 个 worker 并发地读取 Redis，弹出任务
? BRPOP key [key ...] timeout
? redis-rb 是线程安全的，每个命令都是同步过
的（使用MonitorMixin）
? 弹出后处理任务，可以使用 middleware
? lib/sidekiq/manager.rb
? lib/sidekiq/fetch.rb

定时任务
? 使用有序集合保存队列
? ?一个Actor轮询 schedule 有序集合。先用 ZRANGEBYSCORE 从 schedule 查出?一个任务，接着用
ZREM 删除 schedule 中的这个任务，再把该任务使用 LPUSH 放回对应队列的列表（List）。
? ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]
? 有序集合的 score 为执行时刻
? 返回执行时刻在 min 和 max 之间的定时任务（但没有从有序集合删除）
? ZREM key member [member ...]
? 移除有序集合里的成员
? 从有序集合删除定时任务
? lib/sidekiq/scheduled.rb

任务小结
? 读取并从队列（List）中删除元素，是原
子性的操作。使用 List 存取普通任务。
? 使用有序集合（SortedSet）存取定时任
务，先查询，再从有序集合删除，然后
放回对应的普通队列。定时任务并不是
定时执行，而是定时放入普通任务队列
中。

Middleware
? lib/sidekiq/middleware/chain.rb
? 举例
? Sidekiq::Middleware::Server::RetryJobs
? sidekiq-failures
? kiqstand
? 在每个任务完成后断开 mongoid 的连接

链式调用实现（1）
Sidekiq.server_middleware.invoke(worker, msg, queue) do
worker.perform(*cloned(msg['args']))
end

链式调用实现（2）
def invoke(*args, &?nal_action)
chain = retrieve.dup
traverse_chain = lambda do
if chain.empty?
?nal_action.call
else
chain.shift.call(*args, &traverse_chain)
end
end
traverse_chain.call
end

任务重试
? Sidekiq::Middleware::Server::RetryJobs
? lib/sidekiq/middleware/server/retry_jobs.rb
? 使用middleware实现

任务重试
? 有异常后，给任务添加更多的重新信息，然后保存在有序集合 “retry”
中作为定时任务执行（处理方式和定时任务相同）
? 最大重试次数
? retry_attempts_from(msg['retry'],
DEFAULT_MAX_RETRY_ATTEMPTS)
? 在任务中保存重试信息
? retry_count、retried_at 等
? 重试的时间
? seconds_to_delay(count)

任务失败
? 默认重试25次，该任务将不再重试
? 给 worker 定义 retries_exhausted 方法，
在终止重试后执行

任务重试小结
? 重试任务使用名称为 retry 的有序集合，
将失败的任务转换成定时任务

延迟连接 Redis
? 不需要在 fork 后手动重连 Redis
? b0def215e1231745153209a28813b82f98c6bcaa
? http://www.modrails.com/documentation/Users
%20guide
%20Nginx.html#spawning_methods_explained
? 在Rails和项目初始化后创建的连接
（Connection），不需要在fork后重连

多线程下的日志
? Ruby 标准库的 Logger 是线程安全的
? 使用 MonitorMixin
? lib/ruby/1.9.1/logger.rb

其他
? 如何并发的？
? Celluloid

参考资料
? https://github.com/mperham/sidekiq
? https://github.com/celluloid/celluloid
? Working with Ruby Threads
? Working With Unix Processes

谢谢！

狠狠撸

Sidekiq 源码分析

Recommended

More Related Content

Viewers also liked (7)

Similar to Sidekiq 源码分析 (17)

Sidekiq 源码分析