ป้องกัน Sidekiq Jobs ซ้ำ: 3 กลยุทธ์ที่พิสูจน์แล้ว

โอกาสสูงมากที่หากคุณเขียนโค้ด Ruby คุณใช้ Sidekiq จัดการ Background Processing หากคุณมาจาก ActiveJob หรือพื้นฐานอื่นๆ 请继续关注 เนื่องจากเคล็ดลับบางส่วนที่ครอบคลุมสามารถนำไปใช้ได้ที่นั่นเช่นกัน

ผู้คนใช้ Sidekiq Background Jobs สำหรับกรณีต่างๆ บางคนบีบตัวเลข บางคนส่งอีเมลต้อนรับให้ผู้ใช้ และบางคนจัดตารางการซิงค์ข้อมูล 无论กรณีของคุณคืออะไร คุณอาจจะเจอกับข้อกำหนดที่ต้องหลีกเลี่ยง Duplicate Jobs โดย Duplicate Jobs ที่นี่หมายถึง Job 2 ตัวที่ทำสิ่งเดียวกันทุกประการ มาเจาะลึกลงไปหน่อย

ทำไมต้อง De-duplicate Jobs?

จินตนาการสถานการณ์ที่ Job ของคุณมีลักษณะดังนี้:

class BookSalesWorker include Sidekiq::Worker def perform(book_id) crunch_some_numbers(book_id) upload_to_s3 end ... end

BookSalesWorker ทำสิ่งเดียวกันเสมอ — Query DB หาหนังสือจาก book_id และดึงข้อมูล Sales ล่าสุดมาคำนวณตัวเลขบางอย่าง แล้วอัปโหลดไปยังบริการเก็บข้อมูล จำไว้ว่า ทุกครั้งที่มีการขายหนังสือบนเว็บไซต์ของคุณ Job นี้จะถูก Enqueue

ตอนนี้ ถ้าคุณได้ Sales 100 ครั้งพร้อมกัน? คุณจะมี Job 100 ตัวที่ทำสิ่งเดียวกันทุกประการ อาจคุณไม่แคร์กับ S3 Writes มากนัก และ Queue ของคุณไม่ได้ Congested มากนัก จึงรับภาระได้ แต่นั่น "Scale ได้ไหม?"™️

แน่นอนว่าไม่ได้ หากคุณเริ่มรับ Sales มากขึ้นสำหรับหนังสือมากขึ้น Queue ของคุณจะ 쌓ي่งด้วยงานที่ไม่จำเป็นอย่างรวดเร็ว หากคุณมี Job 100 ตัวที่ทำสิ่งเดียวกันสำหรับหนังสือเล่มเดียว และคุณมี 10 หนังสือที่ขายพร้อมกัน ตอนนี้คุณมี Job ลึก 1000 ตัวใน Queue ซึ่งในความเป็นจริง คุณอาจมีแค่ 10 Job สำหรับแต่ละหนังสือ

ตอนนี้ มาดูตัวเลือกกันบ้างว่าคุณสามารถป้องกัน Duplicate Jobs ที่สะสมใน Queue ได้อย่างไร

1. ทำเอง (DIY Way)

หากคุณไม่ชอบ External Dependencies และ Logic ที่ซับซ้อน คุณสามารถเพิ่ม Custom Solution ลงใน Codebase ได้ ฉันสร้าง Repo ตัวอย่างเพื่อลองตัวอย่างของเราก่อน จะมี Link ในแต่ละวิธีการ

1.1 วิธี One Flag

คุณสามารถเพิ่ม Flag หนึ่งตัวที่ตัดสินใจว่าจะ Enqueue Job หรือไม่ คนหนึ่งอาจเพิ่ม sales_enqueued_at ในตาราง Book และดูแลมัน ตัวอย่างเช่น:

module BookSalesService def schedule_with_one_flag(book) # ตรวจสอบว่ Job ถูก Enqueue เมื่อ 10 นาทีที่แล้ว if book.sales_enqueued_at < 10.minutes.ago book.update(sales_enqueued_at: Time.current) BookSalesWorker.perform_async(book.id) end end end

นั่นหมายความว่าไม่มี Job ใหม่จะถูก Enqueue จนกว่าจะผ่านไป 10 นาทีจากเวลาที่ Job ล่าสุดถูก Enqueue หลังจากผ่านไป 10 นาที เราจึง Update sales_enqueued_at และ Enqueue Job ใหม่

อีกวิธีหนึ่งคือตั้ง Flag หนึ่งตัวที่เป็น Boolean เช่น crunching_sales คุณตั้ง crunching_sales เป็น true ก่อนที่ Job แรกจะถูก Enqueue แล้วตั้งเป็น false เมื่อ Job เสร็จสิ้น Job อื่นๆ ที่พยายามจะถูก Scheduled จะถูก Reject จนกว่า crunching_sales จะเป็น false

คุณสามารถลองวิธีนี้ใน Example Repo ที่ฉันสร้างไว้

1.2 วิธี Two Flags

หาก "Lock" Job ไม่ให้ Enqueue เป็นเวลา 10 นาที聽起來น่ากลัวเกินไป แต่คุณยังยอมรับ Flag เพิ่มเติมในโค้ดได้ วิธีต่อไปอาจน่าสนใจคุณ

คุณสามารถเพิ่ม Flag อีกตัวเข้าไปใน sales_enqueued_at ที่มีอยู่ — คือ sales_calculated_at โค้ดของเราจะมีลักษณะประมาณนี้:

module BookSalesService def schedule_with_two_flags(book) # ตรวจสอบว่ายอดขายกำลังถูกคำนวณอยู่หรือไม่ if book.sales_enqueued_at <= book.sales_calculated_at book.update(sales_enqueued_at: Time.current) BookSalesWorker.perform_async(book.id) end end end class BookSalesWorker include Sidekiq::Worker def perform(book_id) crunch_some_numbers(book_id) upload_to_s3 # เพิ่มใหม่ book.update(sales_calculated_at: Time.current) end ... end

หากต้องการลอง ให้ดูคำแนะนำใน Example Repo

ตอนนี้เราควบคุมช่วงเวลาระหว่างที่ Job ถูก Enqueue และเสร็จสิ้น ในช่วงเวลานั้น ไม่มี Job ใดจะถูก Enqueue ขณะที่ Job กำลังรัน sales_enqueued_at จะมากกว่า sales_calculated_at เมื่อ Job รันเสร็จ sales_calculated_at จะใหม่กว่า (Recent มากกว่า) sales_enqueued_at และ Job ใหม่จะถูก Enqueue

การใช้ Two Flags อาจน่าสนใจ เพราะคุณสามารถแสดงเวลาล่าสุดที่ตัวเลข Sales ได้รับการ Update ใน UI ได้ แล้วผู้ใช้อ่านข้อมูลสามารถรู้ได้ว่าข้อมูลมี多近期 Win-win situation

สรุป Flag

อาจน่าหลงใหลที่จะสร้าง Solution แบบนี้ในช่วงที่ต้องการ แต่ในมุมมองของฉัน มันดูค่อนข้าง Clumsy และเพิ่ม Overhead บ้าง ฉันแนะนำให้ใช้ถ้า Use Case ของคุณเรียบง่าย แต่ทันทีที่มันพิสูจน์ว่าซับซ้อนหรือไม่พอ ฉันจะร้องขอให้คุณลองตัวเลือกอื่น

ข้อเสียใหญ่ของ Flag Approach คือคุณจะสูญเสีย Job ทั้งหมดที่พยายาม Enqueue ระหว่าง 10 นาทีนั้น ข้อดีใหญ่คือคุณไม่ได้นำ Dependencies เข้ามา และจะช่วยลดจำนวน Job ใน Queue ได้ค่อนข้างรวดเร็ว

1.3 การ Traverse Queue

อีกวิธีหนึ่งคือการสร้าง Custom Locking Mechanism เพื่อป้องกัน Job เดียวกันจากการ Enqueue เราจะตรวจสอบ Sidekiq Queue ที่เราสนใจ และดูว่า Job (Worker) อยู่ใน Queue แล้วหรือยัง โค้ดจะมีลักษณะประมาณนี้:

module BookSalesService def schedule_unique_across_queue(book) queue = Sidekiq::Queue.new('default') queue.each do |job| return if job.klass == BookSalesWorker.to_s && job.args == [book.id] end BookSalesWorker.perform_async(book.id) end end class BookSalesWorker include Sidekiq::Worker def perform(book_id) crunch_some_numbers(book_id) upload_to_s3 end ... end

ในตัวอย่างข้างต้น เราตรวจสอบว่า Queue 'default' มี Job ที่มี Class Name เป็น BookSalesWorker หรือไม่ เรายังตรวจสอบว่า Job Arguments ตรงกับ Book ID หรือไม่ หาก Job BookSalesWorker พร้อม Book ID เดียวกันอยู่ใน Queue เราจะ Return ก่อนเวลาและไม่ Schedule อีกตัว

โปรดทราบว่าบางตัวอาจถูก Scheduled หากคุณ Schedule Jobs เร็วเกินไปเพราะ Queue ว่างเปล่า สิ่งเดียวกันเกิดขึ้นกับฉันตอนทดสอบใน Local ด้วย:

100.times { BookSalesService.schedule_unique_across_queue(book) }

คุณสามารถลองได้ใน Example Repo

ข้อดีของวิธีนี้คือคุณสามารถ Traverse ทุก Queue เพื่อค้นหา Job ที่มีอยู่แล้วหากคุณต้องการ ข้อเสียคือคุณยังอาจมี Duplicate Jobs หาก Queue ว่างและคุณ Schedule มากเกินไปพร้อมกัน นอกจากนี้ คุณอาจ Traverse ผ่านทุก Job ใน Queue ก่อน Schedule หนึ่งตัว ดังนั้นอาจมีค่าใช้จ่ายสูงขึ้นอยู่กับขนาด Queue ของคุณ

2. อัปเกรดเป็น Sidekiq Enterprise

หากคุณหรือองค์กรของคุณมีเงินเหลืออยู่บ้าง คุณสามารถอัปเกรดเป็น Enterprise Version ของ Sidekiq เริ่มต้นที่ $179 ต่อเดือน และมีฟีเจอร์สุดโปลที่ช่วยหลีกเลี่ยง Duplicate Jobs น่าเสียดายที่ฉันไม่มี Sidekiq Enterprise แต่ฉันเชื่อว่า Documentation ของพวกเขาเพียงพอ คุณสามารถมี Unique (Non-duplicated) Jobs ได้ง่ายด้วยโค้ดต่อไปนี้:

class BookSalesWorker include Sidekiq::Worker sidekiq_options unique_for: 10.minutes def perform(book_id) crunch_some_numbers(book_id) upload_to_s3 end ... end

และแค่นั้นแหละ คุณมี Implementation Job คล้ายกับที่เราอธิบายในส่วน 'One Flag Approach' Job จะ Unique เป็นเวลา 10 นาที หมายความว่า Job อื่นๆ ที่มี Arguments เดียวกันไม่สามารถ Scheduled ได้ในช่วงเวลานั้น

One-liner ที่ค่อนข้าง Cool หน่ะ? หากคุณมี Enterprise Sidekiq และเพิ่งรู้ฟีเจอร์นี้ ฉันยินดีมากที่ช่วยได้ แต่เราส่วนใหญ่ไม่ได้ใช้ มาไปต่อที่ Solution ถัดไปกัน

3. sidekiq-unique-jobs ช่วยชีวิต

ใช่ ฉันรู้ว่าเรากำลังจะพูดถึง Gem และใช่ มันมีไฟล์ Lua บ้างซึ่งอาจทำให้บางคนกลัว แต่ терพ์ใจไปก่อน มันเป็น Deal ที่ Sweet จริงๆ ที่คุณได้รับ Gem sidekiq-unique-jobs มาพร้อม Locking และ Configuration Options มากมาย — อาจมากกว่าที่คุณต้องการ

เพื่อเริ่มต้นอย่างรวดเร็ว ใส่ Gem sidekiq-unique-jobs ลงใน Gemfile ทำ bundle และกำหนดค่า Worker ตามที่แสดง:

class UniqueBookSalesWorker include Sidekiq::Worker sidekiq_options lock: :until_executed, on_conflict: :reject def perform(book_id) book = Book.find(book_id) logger.info "I am a Sidekiq Book Sales worker - I started" sleep 2 logger.info "I am a Sidekiq Book Sales worker - I finished" book.update(sales_calculated_at: Time.current) book.update(crunching_sales: false) end end

มี Options มากมาย แต่ฉันตัดสินใจทำให้ Simple และใช้ตัวนี้:

sidekiq_options lock: :until_executed, on_conflict: :reject

lock: :until_executed จะ Lock Job UniqueBookSalesWorker แรกจนกว่าจะ Executed ด้วย on_conflict: :reject เราบอกว่าต้องการให้ Job อื่นๆ ที่พยายาม Execute ถูก Reject เข้า Dead Queue สิ่งที่เราทำได้คล้ายกับที่เราทำใน DIY Examples ข้างต้น

จุดที่ดีกว่า DIY Examples เล็กน้อยคือเรามี Log ประมาณว่าอะไรเกิดขึ้น เพื่อให้เห็นภาพ ลองทำดู:

5.times { UniqueBookSalesWorker.perform_async(Book.last.id) }

จะมีเพียง 1 Job ที่ Execute เสร็จสมบูรณ์ และอีก 4 Job จะถูกส่งไป Dead Queue ซึ่งคุณสามารถ Retry ได้ วิธีนี้แตกต่างจากตัวอย่างของเราที่ Duplicate Jobs ถูกเพิกเฉย

มี Options มากมายสำหรับ Locking และ Conflict Resolving ฉันแนะนำให้คุณอ่าน Documentation ของ Gem สำหรับ Use Case เฉพาะของคุณ

ข้อมูลเชิงลึกที่ยอดเยี่ยม

สิ่งที่ยอดเยี่ยมของ Gem นี้คือคุณสามารถดู Locks และ History ของสิ่งที่เกิดขึ้นใน Queue ได้ ทั้งที่คุณต้องทำคือเพิ่มบรรทัดต่อไปนี้ใน config/routes.rb:

# config/routes.rb require 'sidekiq_unique_jobs/web' Rails.application.routes.draw do mount Sidekiq::Web, at: '/sidekiq' end 

มันจะ Include Sidekiq Client ต้นฉบับ แต่ยังให้คุณ 2 หน้าเพิ่ม — หน้าสำหรับ Job Locks และอีกหน้าสำหรับ Changelog มันจะมีลักษณะแบบนี้:

สังเกตว่าเรามี 2 หน้าใหม่ "Locks" และ "Changelogs" ฟีเจอร์สุดโปล

คุณสามารถลองทั้งหมดนี้ใน Example Project ที่ Gem ติดตั้งและพร้อมใช้งานแล้ว

ทำไมถึง Lua?

ค่อนข้างแรกที่เห็น Gem ฉันสงสัย: ทำไมต้องใช้ Lua ภายใน Ruby Gem? อาจดูแปลกตอนแรก แต่ Redis รองรับการรัน Lua Scripts ฉันเดาว่า Author ของ Gem น่าจะมีสิ่งนี้ในใจ และต้องการทำ Logic ที่คล่องตัวกว่าใน Lua

หากคุณดูไฟล์ Lua ใน Repo ของ Gem พวกมันไม่ได้ซับซ้อนนัก Lua Scripts ทั้งหมดถูกเรียกภายหลังจาก Ruby Code ใน SidekiqUniqueJobs::Script::Caller ที่นี่ โปรดดู Source Code มันน่าสนใจที่จะอ่านและเข้าใจว่าสิ่งต่างๆ ทำงานอย่างไร

Gem ทางเลือก

หากคุณใช้ ActiveJob อย่างหนัก คุณสามารถลอง Gem active-job-uniqueness ได้ที่นี่ แนวคิดคล้ายกัน แต่แทนที่จะใช้ Custom Lua Scripts มันใช้ [Redlock] เพื่อ Lock Items ใน Redis

เพื่อให้มี Unique Job ด้วย Gem นี้ คุณสามารถจินตนาการ Job แบบนี้:

class BookSalesJob < ActiveJob::Base unique :until_executed def perform ... end end

Syntax น้อยกว่าแตคล้มคล้ายกับ Gem sidekiq-unique-jobs อาจแก้กรณีของคุณได้หากคุณพึ่งพา ActiveJob มาก

ความคิดเห็นสุดท้าย

หวังว่าคุณจะได้ความรู้เกี่ยวกับวิธีจัดการ Duplicate Jobs ในแอปของคุณ ฉันสนุกมากกับการ Research และเล่นกับ Solution ต่างๆ หากคุณไม่พบสิ่งที่คุณมองหา ฉันหวังว่าตัวอย่างบางส่วนจะ Inspire ให้คุณสร้างสิ่งของตัวเอง

นี่คือ Example Project พร้อม Code Snippets ทั้งหมด

เจอกันในครั้งหน้า Cheers!

P.S. หากคุณต้องการอ่านโพสต์ Ruby Magic ทันทีที่เผยแพร่ สมัครรับข่าวสาร Ruby Magic newsletter และจะไม่พลาดโพสต์เด็ดๆ ไปเลย!