非同期処理 — ピークを平準化する
同期処理の罠
「投稿ボタンを押してから3秒待たされる」
Buzzのユーザーフィードバックが続々と届いていた。特に投稿作成のレスポンスタイムが平均2.8秒かかっていた。ユーザーはボタンを押して、3秒間じっと待つ。これは体験として最悪だ。
「何をそんなに時間がかかっているのか」とアキラは調べた。
# 問題のある投稿作成処理（同期）
def create
@post = Post.create!(post_params) # 50ms
# 以下が全て同期で実行される（ユーザーが待つ！）
ImageResizeService.call(@post) # 1,200ms ← 画像リサイズ
NotificationService.notify_followers(@post) # 800ms ← 全フォロワーへ通知
HashtagIndexService.index(@post) # 300ms ← OpenSearchにインデックス
SpamDetectionService.check(@post) # 500ms ← AIによるスパム検知
redirect_to @post
# 合計: 2,850ms
end問題は明らかだった。ユーザーに必要なのは「投稿が保存された」という確認だけだ。画像リサイズも、通知送信も、インデックス更新も——全部後回しにできる。
Loading diagram...
INFO
非同期処理の原則: ユーザーのレスポンスに必要なことだけ同期で行う。それ以外は全てバックグラウンドに移す。目安はレスポンスタイム200ms以内。「保存された」という確認は即座にできる。残りはあとでやればいい。
Sidekiqで基本的な非同期処理
RubyのバックグラウンドジョブにはSidekiqが定番だ。Redisをキューとして使い、ワーカープロセスがジョブを順番に処理する。
# Gemfile
gem 'sidekiq'
gem 'sidekiq-pro' # 本番向け（バッチ、スケジューリング、メトリクス）
gem 'sidekiq-cron' # cronスタイルのスケジュールジョブ# config/sidekiq.yml
:concurrency: 10 # ワーカーの並列実行数（DBプール数と要整合）
:timeout: 25 # ジョブのタイムアウト（秒）
:queues:
- [critical, 10] # 優先度高（通知、決済）
- [default, 5] # 通常のジョブ
- [low, 2] # 低優先度（バッチ、集計）
- [mailers, 3] # メール送信
:max_retries: 5 # 失敗時の最大リトライ回数
:dead_max_jobs: 10000 # デッドジョブの最大保持数
# スケジュールジョブ
:schedule:
trending_update:
cron: '*/15 * * * *'
class: UpdateTrendingHashtagsJob
queue: low
daily_digest:
cron: '0 8 * * *'
class: DailyDigestJob
queue: low
cleanup_sessions:
cron: '0 2 * * 0'
class: CleanupOldSessionsJob
queue: low# app/jobs/image_resize_job.rb
class ImageResizeJob < ApplicationJob
queue_as :default
# エラー時の自動リトライ（指数バックオフ）
retry_on StandardError, wait: :exponentially_longer, attempts: 5
# 対象レコードが削除されていたら無視
discard_on ActiveRecord::RecordNotFound
def perform(post_id)
post = Post.find(post_id)
return if post.image.blank?
# 3つのサイズにリサイズ
variants = [
{ size: [150, 150], suffix: 'thumb', quality: 80 },
{ size: [600, 600], suffix: 'medium', quality: 85 },
{ size: [1200, 1200], suffix: 'large', quality: 90 }
]
variants.each do |variant|
post.image.variant(
resize_to_limit: variant[:size],
format: :webp,
quality: variant[:quality]
).processed # 事前生成してS3にキャッシュ
end
post.update_columns(
image_processed: true,
image_processed_at: Time.current
)
Rails.logger.info "ImageResizeJob: post_id=#{post_id} processed successfully"
end
end# app/jobs/notify_followers_job.rb
class NotifyFollowersJob < ApplicationJob
queue_as :critical # 通知は優先度高
def perform(post_id)
post = Post.includes(:user).find(post_id)
followers = post.user.followers
.where(notification_settings: { new_post: true })
.select(:id, :push_token)
# フォロワー数が多い場合はバッチ処理
if followers.count > 10_000
# 大量フォロワー: 1000件ずつに分割して処理
followers.find_in_batches(batch_size: 1000) do |batch|
PushNotificationBatchJob.perform_later(
post_id: post.id,
follower_ids: batch.map(&:id)
)
end
else
# 通常: 直接送信
followers.each do |follower|
PushNotificationJob.perform_later(
recipient_id: follower.id,
post_id: post.id,
message: "#{post.user.username}が新しい投稿をしました"
)
end
end
end
end# app/jobs/push_notification_job.rb
class PushNotificationJob < ApplicationJob
queue_as :critical
retry_on Expo::Error, wait: 10.seconds, attempts: 3
def perform(recipient_id:, post_id:, message:)
recipient = User.find(recipient_id)
return unless recipient.push_token.present?
return unless recipient.notifications_enabled?
# Expo Push Notification（React Native向け）
client = Expo::Client.new
client.publish(
to: recipient.push_token,
title: 'Buzz',
body: message,
data: { type: 'new_post', post_id: post_id }
)
# 通知ログを記録
Notification.create!(
recipient_id: recipient_id,
post_id: post_id,
type: 'new_post',
delivered_at: Time.current
)
end
end改善後の投稿作成処理
# app/controllers/posts_controller.rb（改善後）
def create
@post = Post.create!(post_params) # 50ms のみ同期
# 残りは全て非同期（即座にキューに積まれる）
ImageResizeJob.perform_later(@post.id)
NotifyFollowersJob.perform_later(@post.id)
HashtagIndexJob.perform_later(@post.id)
SpamDetectionJob.perform_later(@post.id)
# 即座にレスポンスを返す
redirect_to @post, notice: '投稿しました！'
# 合計: 55ms（前: 2,850ms → 51倍改善！）
end改善効果:
投稿作成レスポンスタイム: 2,850ms → 55ms（51倍改善）
ユーザーが待つ時間: 大幅短縮
バックグラウンド処理: 5秒以内に完了（ユーザーには見えない）
エラー時: リトライで自動復旧（ユーザーへの影響ゼロ）
AWS SQS との統合
Sidekiqは便利だが、Redisに依存している。Redisが落ちるとジョブが失われる可能性がある。大規模なシステムではAWS SQSとの統合が信頼性を高める。
Loading diagram...
# Gemfile
gem 'shoryuken' # SQS をバックエンドにする Sidekiq 互換ライブラリ
# または
gem 'aws-sdk-sqs' # SQS 直接操作# config/shoryuken.yml
aws:
region: ap-northeast-1
queues:
- buzz-critical # 優先度高
- buzz-default # 通常
- buzz-email # メール
concurrency: 20
delay: 0
timeout: 60 # 60秒でジョブがタイムアウト# SQS を直接使うサービス
# app/services/sqs_publisher.rb
class SqsPublisher
SQS = Aws::SQS::Client.new(region: 'ap-northeast-1')
QUEUES = {
critical: ENV['SQS_CRITICAL_QUEUE_URL'],
default: ENV['SQS_DEFAULT_QUEUE_URL'],
email: ENV['SQS_EMAIL_QUEUE_URL']
}.freeze
def self.publish(queue:, job_class:, **args)
SQS.send_message(
queue_url: QUEUES.fetch(queue),
message_body: {
job_class: job_class.to_s,
args: args,
enqueued_at: Time.current.iso8601
}.to_json,
# FIFOキューの場合: メッセージグループIDで順序を保証
# message_group_id: args[:user_id].to_s,
# message_deduplication_id: SecureRandom.uuid,
delay_seconds: 0
)
end
endWARNING
SQSのat-least-once deliveryに注意。同じメッセージが複数回配送される可能性がある。ジョブは冪等（べきとう、何度実行しても同じ結果になる）に実装する。「メールが2回送られた」「いいね数が2回増えた」などの問題を防ぐ。
# 冪等なジョブの実装パターン
# パターン1: DBのuniqueインデックスで重複を防ぐ
class ProcessPaymentJob < ApplicationJob
def perform(order_id:, idempotency_key:)
# 冪等性キーで重複チェック
return if PaymentRecord.exists?(idempotency_key: idempotency_key)
order = Order.find(order_id)
payment = PaymentGateway.charge(
amount: order.total_amount,
idempotency_key: idempotency_key # 決済APIも冪等キーを送る
)
PaymentRecord.create!(
order_id: order_id,
payment_id: payment.id,
idempotency_key: idempotency_key,
charged_at: Time.current
)
end
end
# パターン2: フラグで実行済みを管理
class SendWelcomeEmailJob < ApplicationJob
def perform(user_id)
user = User.find(user_id)
# 既に送信済みならスキップ（冪等）
return if user.welcome_email_sent_at.present?
WelcomeMailer.send_welcome(user).deliver_now
# 送信済みフラグを立てる（排他ロックで競合防止）
user.with_lock do
user.update!(welcome_email_sent_at: Time.current) if user.welcome_email_sent_at.nil?
end
end
end
# パターン3: Redisで重複実行をブロック
class UpdateUserStatsJob < ApplicationJob
REDIS = Redis.new(url: ENV['REDIS_URL'])
LOCK_TTL = 10.minutes.to_i
def perform(user_id)
lock_key = "job:update_user_stats:#{user_id}"
# Redis で分散ロックを取得
acquired = REDIS.set(lock_key, Process.pid, nx: true, ex: LOCK_TTL)
unless acquired
Rails.logger.info "UpdateUserStatsJob: skipping duplicate for user #{user_id}"
return
end
begin
user = User.find(user_id)
user.update!(
posts_count: user.posts.count,
followers_count: user.followers.count,
stats_updated_at: Time.current
)
ensure
REDIS.del(lock_key)
end
end
endスケジュールジョブ
定期的なバッチ処理もバックグラウンドで行う。
# app/jobs/daily_digest_job.rb
class DailyDigestJob < ApplicationJob
queue_as :low
def perform
Rails.logger.info "DailyDigestJob: started at #{Time.current}"
# 前日アクティブだったユーザーに日次ダイジェストを送る
target_users = User.where(
last_active_at: 2.days.ago..1.day.ago,
digest_enabled: true
)
total = target_users.count
Rails.logger.info "DailyDigestJob: processing #{total} users"
processed = 0
target_users.find_each(batch_size: 500) do |user|
DigestMailer.daily_digest(user).deliver_later(queue: :mailers)
processed += 1
# 進捗ログ（大量処理の場合）
Rails.logger.info "DailyDigestJob: #{processed}/#{total}" if processed % 1000 == 0
end
Rails.logger.info "DailyDigestJob: completed. #{processed} emails queued."
end
end# app/jobs/update_trending_hashtags_job.rb
class UpdateTrendingHashtagsJob < ApplicationJob
queue_as :low
REDIS = Redis.new(url: ENV['REDIS_URL'])
def perform
# 直近15分の投稿からハッシュタグを集計
recent_posts = Post.where(created_at: 15.minutes.ago..)
.pluck(:content)
hashtag_counts = Hash.new(0)
recent_posts.each do |content|
content.scan(/#\w+/).each do |tag|
hashtag_counts[tag.downcase] += 1
end
end
# Redis の Sorted Set に保存
hashtag_counts.each do |tag, count|
REDIS.zadd('trending:hashtags', count, tag)
end
# 上位100件以外を削除（メモリ節約）
REDIS.zremrangebyrank('trending:hashtags', 0, -101)
REDIS.expire('trending:hashtags', 30.minutes.to_i)
end
endキューの優先順位と Dead Letter Queue
# app/jobs/application_job.rb
class ApplicationJob < ActiveJob::Base
# 一般的なエラーへの対応
retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3
retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 5
retry_on Sidekiq::Shutdown, wait: 10.seconds, attempts: 5
# 致命的なエラーは即座に失敗
discard_on ActiveRecord::RecordNotFound # レコードが削除されていたら諦める
discard_on ActiveRecord::RecordInvalid # バリデーションエラーはリトライ不要
# ジョブの実行時間を計測
around_perform do |job, block|
start = Time.current
block.call
duration = Time.current - start
if duration > 30 # 30秒以上かかったジョブを警告
Rails.logger.warn "[SLOW JOB] #{job.class.name} took #{duration.round(2)}s " \
"args=#{job.arguments.inspect}"
end
# メトリクスを記録
StatsD.histogram('sidekiq.job.duration', duration * 1000,
tags: ["job:#{job.class.name}", "queue:#{job.queue_name}"])
end
# ジョブ開始・終了をログ
before_perform do |job|
Rails.logger.info "[JOB START] #{job.class.name} job_id=#{job.job_id}"
end
after_perform do |job|
Rails.logger.info "[JOB DONE] #{job.class.name} job_id=#{job.job_id}"
end
end# AWS SQS Dead Letter Queue 設定（CloudFormation）
BuzzCriticalQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: buzz-critical
VisibilityTimeout: 60 # 処理に60秒まで
MessageRetentionPeriod: 86400 # メッセージ保持: 1日
ReceiveMessageWaitTimeSeconds: 20 # Long Polling（コスト削減）
RedrivePolicy:
deadLetterTargetArn: !GetAtt BuzzCriticalDLQ.Arn
maxReceiveCount: 5 # 5回失敗したらDLQへ移動
BuzzCriticalDLQ:
Type: AWS::SQS::Queue
Properties:
QueueName: buzz-critical-dlq
MessageRetentionPeriod: 1209600 # 14日間保持（調査用）
# DLQにメッセージが届いたらSlackに通知
DLQAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
MetricName: ApproximateNumberOfMessagesVisible
Namespace: AWS/SQS
Statistic: Sum
Period: 60
Threshold: 1
AlarmActions: [!Ref SlackNotificationTopic]
Dimensions:
- Name: QueueName
Value: buzz-critical-dlqSidekiqのモニタリング
# config/routes.rb（Sidekiqダッシュボードを保護）
require 'sidekiq/web'
Rails.application.routes.draw do
authenticate :user, ->(u) { u.admin? } do
mount Sidekiq::Web => '/sidekiq'
end
end
# Sidekiq::Web の Basic認証（APIキーで保護）
Sidekiq::Web.use Rack::Auth::Basic do |username, password|
ActiveSupport::SecurityUtils.secure_compare(
username, ENV['SIDEKIQ_USERNAME']
) & ActiveSupport::SecurityUtils.secure_compare(
password, ENV['SIDEKIQ_PASSWORD']
)
end# CloudWatch へのメトリクス送信
# app/jobs/sidekiq_metrics_job.rb
class SidekiqMetricsJob < ApplicationJob
queue_as :low
def perform
stats = Sidekiq::Stats.new
cloudwatch = Aws::CloudWatch::Client.new
cloudwatch.put_metric_data({
namespace: 'Buzz/Sidekiq',
metric_data: [
{ metric_name: 'EnqueuedJobs', value: stats.enqueued, unit: 'Count' },
{ metric_name: 'ProcessedJobs', value: stats.processed, unit: 'Count' },
{ metric_name: 'FailedJobs', value: stats.failed, unit: 'Count' },
{ metric_name: 'DeadJobs', value: stats.dead_size, unit: 'Count' },
{ metric_name: 'WorkerCount', value: stats.workers_size, unit: 'Count' }
]
})
# 各キューの深さを記録
Sidekiq::Queue.all.each do |queue|
cloudwatch.put_metric_data({
namespace: 'Buzz/Sidekiq',
metric_data: [{
metric_name: 'QueueDepth',
value: queue.size,
unit: 'Count',
dimensions: [{ name: 'QueueName', value: queue.name }]
}]
})
end
end
endSidekiqのスケーリング
Sidekiqのワーカー自体もECS Auto Scalingでスケールさせる。
# ECS サービス: Sidekiqワーカー専用
SidekiqService:
Type: AWS::ECS::Service
Properties:
Cluster: !Ref BuzzCluster
TaskDefinition: !Ref SidekiqTaskDef
DesiredCount: 2
# Sidekiq ワーカーのAuto Scaling
SidekiqScalingTarget:
Type: AWS::ApplicationAutoScaling::ScalableTarget
Properties:
MaxCapacity: 20 # 最大20ワーカー
MinCapacity: 2 # 最小2ワーカー
# SQSキューの深さでスケール
SidekiqScalingBySQS:
Type: AWS::ApplicationAutoScaling::ScalingPolicy
Properties:
PolicyType: TargetTrackingScaling
TargetTrackingScalingPolicyConfiguration:
CustomizedMetricSpecification:
MetricName: ApproximateNumberOfMessagesVisible
Namespace: AWS/SQS
Statistic: Sum
Dimensions:
- Name: QueueName
Value: buzz-critical
TargetValue: 100 # ワーカー1台あたり100メッセージをターゲット負荷平準化の効果
Loading diagram...
| 指標 | 改善前 | 改善後 | 改善率 |
|---|---|---|---|
| 投稿作成レスポンス | 2,850ms | 55ms | 51倍 |
| ピーク処理能力 | 200 RPS | 2,000 RPS | 10倍 |
| メール送信エラー率 | 8.3% | 0.01% | 830倍 |
| 画像処理の失敗率 | 12% | 0.1% | 120倍 |
「バックグラウンド処理に移したことで、APIが51倍速くなった」ユイが言った。「でも投稿の画像、まだ表示が遅い。CDNを入れていないから」
次章では、CloudFrontと静的アセット最適化で、世界中どこでも高速表示を実現する方法を学ぶ。
付録: 非同期処理設計のチェックリスト
ジョブ設計:
□ ジョブは冪等か（何度実行しても同じ結果になるか）
□ 対象レコードが削除された場合のハンドリングがあるか
□ タイムアウトを設定しているか（無限に実行されない）
□ リトライ回数と間隔は適切か
キュー設計:
□ 優先度に応じてキューを分けているか
□ Dead Letter Queue を設定しているか
□ DLQにメッセージが入ったらアラートが出るか
監視:
□ キューの深さをモニタリングしているか
□ ジョブの失敗率をモニタリングしているか
□ 処理時間の遅いジョブに警告が出るか
□ Sidekiqダッシュボードへのアクセス制御ができているか
セキュリティ:
□ ジョブの引数に機密情報を含めていないか
（パスワード、トークンなど → IDだけ渡してDB取得）
□ ジョブの実行結果をログに出力する際にPIIが含まれていないか