Upgrade to Pro — share decks privately, control downloads, hide ads and more …

PerfectQueueはいかにパーフェクトか、あるいはRubyとMySQLでジョブキューを作...

 PerfectQueueはいかにパーフェクトか、あるいはRubyとMySQLでジョブキューを作る試みについて

NARUSE, Yui

July 14, 2016
Tweet

More Decks by NARUSE, Yui

Other Decks in Programming

Transcript

  1. 1FSGFDU2VFVFͷಛ௃ w "UMFBTUPODFTFNBOUJDT w 1FSNBOFOU4UPSBHF w 3%#.4 .Z42-  w

    SFUSZ IFBSUCFBUUJNFPVU  w "UNPTUPODF  w *EFNQPUFOUUBTLTVCNJTTJPO w 1FSGFDU2VFVF`TVOJRVFLFZ w 0UIFSVOJRVFOFTTPO"QQMJDBUJPO-BZFS USBEFP⒎
  2. 5IFMJGFPG UBTLT w 4VCNJU w "DRVJSF w 'JOJTI w %FMFUF

    TVCNJU BDRVJSF EFMFUF GJOJTI SFMFBTF
  3. %BUB4USVDUVSF CREATE TABLE `queue` (
 /* unique key (" at

    most once) */
 id VARCHAR(255) NOT NULL,
 /* for FIFO’s timeline */
 timeout INT NOT NULL,
 /* opaque data */
 data LONGBLOB NOT NULL,
 /* alive or finished */
 created_at INT,
 PRIMARY KEY (id)
 )
  4. λεΫͷ౤ೖ INSERT INTO `queue`
 (id, timeout, data, created_at)
 VALUES (


    ‘hive.123456789.unique_key’, /* unique key */
 UNIX_TIMESTAMP(), /* now */
 ’{“some”: “json”}’, /* opaque data */
 NOW()); /* NOT NULL */ UJNFPVU now JOTFSU
  5. /* Run the following 2 queries exclusively: */ SELECT id

    FROM `queue`
 WHERE timeout < UNIX_TIMESTAMP()
 AND created_at IS NOT NULL /* ະ׬ྃͷλεΫ */
 ORDER BY timeout ASC LIMIT 5; UPDATE SET timeout=UNIX_TIMESTAMP()+300 /* alive_time */
 WHERE id IN (?,?,?,?,?); λεΫͷऔಘ UJNFPVU now BDRVJSF +300 waiting... running...
  6. UPDATE `queue`
 SET timeout = UNIX_TIMESTAMP()+300
 WHERE id = ?;

    λεΫͷ࣮ߦதʜʜ UJNFPVU now IFBSUCFBU +300 waiting... running... )FBSUCFBUEVSJOHSVOOJOHBUBTLʜ
  7. λεΫͷ׬ྃ UPDATE `queue`
 SET timeout = UNIX_TIMESTAMP() + 720,
 created_at

    = NULL UJNFPVU now pOJTI +720 ˢ SFUFOUJPO@UJNF running... +300 ˞SFUFOUJPO@UJNF
 ɹ׬ྃͯ͠΋ॏෳ࣮ߦΛ๷͙ͨΊɺ
 ɹҰఆ࣌ؒ෺ཧ࡟আͤͣʹอ͓࣋ͯ͘͠
  8. ׬ྃࡁΈλεΫͷ࡟আ DELETE FROM `queue`
 WHERE timeout <= UNIX_TIMESTAMP()
 AND created_at

    IS NULL; UJNFPVU now EFMFUF +720 to be deleted... લड़ͷSFUFOUJPO@UJNFΛա͗ͨΒ෺ཧ࡟আɻ ىಈ௚ޙͱҰఆ਺λεΫΛऔಘ͢Δ͝ͱʹ
 ্هΫΤϦΛ౤͛Δɻ
  9. λεΫऔಘͲ͏ഉଞ͢Δ͔ w -0$,5"#-&4CMPDLTPUIFSQSPDFTTOPUPOMZ BDRVJSJOHUBTLTCVUBMTPTVCNJUUJOHUBTLT */4&35 BOEIFBSUCFBUT ʙɹ-0$,5"#-&4ɹʙ SET autocommit=0; LOCK

    TABLES `queue` WRITE; SELECT id FROM `queue` WHERE … LIMIT 5; UPDATE SET … WHERE id IN (?,?,?,?,?); COMMIT; UNLOCK TABLES;
  10. λεΫऔಘͲ͏ഉଞ͢Δ͔ w (&5@-0$, VTFSMPDL XPSLTpOF ʙɹ(&5@-0$,ɹʙ GET_LOCK(`queue`); SELECT id FROM

    `queue` WHERE … LIMIT 5; UPDATE SET … WHERE id IN (?,?,?,?,?); RELEASE_LOCK(`queue`);
  11. ׬ྃࡁΈλεΫͷ࡟আW GET_LOCK(`queue`); DELETE FROM `queue`
 WHERE timeout <= UNIX_TIMESTAMP()
 AND

    created_at IS NULL; RELEASE_LOCK(`queue`); UJNFPVU now EFMFUF +720 to be deleted... 3VOEFMFUFRVFSZBGUFSUIFMPDLOBNFEUIFUBCMF લड़ͷSFUFOUJPO@UJNFΛա͗ͨΒ෺ཧ࡟আɻ ىಈ௚ޙͱҰఆ਺λεΫΛऔಘ͢Δ͝ͱʹ
 ͜ͷΫΤϦΛ౤͛Δɻ
  12. ׬ྃࡁΈλεΫͷ࡟আW GET_LOCK(`queue`); DELETE FROM `queue`
 WHERE timeout <= UNIX_TIMESTAMP()
 AND

    created_at IS NULL; RELEASE_LOCK(`queue`); UJNFPVU now EFMFUF +720 to be deleted... લड़ͷSFUFOUJPO@UJNFΛա͗ͨΒ෺ཧ࡟আɻ ىಈ௚ޙͱҰఆ਺λεΫΛऔಘ͢Δ͝ͱʹ
 ͜ͷΫΤϦΛ౤͛Δɻ શͯͷϫʔΧʔΛಉ࣌ʹىಈ͢Δͱɺ
 ͠͹Β͘ͷؒ࡟আΫΤϦ͕ඈ͹ͳ͍
  13. /FUXPSLMBUFODZ w (&5@-0$, VTFSMPDL XPSLTpOF
 XIJMFOFUXPSLJTpOF w &$ͱ3%4ͷωοτϫʔΫ஗Ԇ͕ѱԽ͢Δͱɺ
 εϧʔϓοτ͕ഒѱԽ͢Δ ʙɹBDRVJSJOHUBTLTXJUI(&5@-0$,

    ɹʙ GET_LOCK(`queue`); SELECT id FROM `queue` WHERE … LIMIT 5; UPDATE SET … WHERE id IN (?,?,?,?,?); RELEASE_LOCK(`queue`); -0$,
 355
  14. %BUB4USVDUVSF CREATE TABLE `queue` (
 /* unique key (" at

    most once) */
 id VARCHAR(255) NOT NULL,
 /* for FIFO’s timeline */
 timeout INT NOT NULL,
 /* opaque data */
 data LONGBLOB NOT NULL,
 /* alive or finished */
 created_at INT,
 /* Who get this task */
 owner BIGINT(21) UNSIGNED NOT NULL DEFAULT 0,
 PRIMARY KEY (id)
 )
  15. UPDATE `queue`
 JOIN (
 SELECT id
 FROM `queue` FORCE INDEX

    (`index_queue_on_timeout`)
 WHERE 1300000000<timeout AND timeout<=UNIX_TIMESTAMP()
 ORDER BY timeout ASC
 LIMIT :max_acquire FOR UPDATE) AS t1 USING(id)
 SET timeout=:next_timeout, owner=CONNECTION_ID() SELECT id, timeout, data, created_at, resource
 FROM `#{@table}`
 WHERE timeout = ? AND owner=CONNECTION_ID() "DRVJSJOHUBTLTXJUIPVUUBCMFMPDLJOH 4FMFDU.BSL ˣ .BSL4FMFDU 28IBUJT$0//&$5*0/@*%  "1SPDFTT*%PGUIF.Z42-DPOOFDUJPO 28IZJUVTFT4&-&$5JO61%"5&  ".Z42-EPFTO`UPQUJNJ[F61%"5&RVFSZXJUI -*.*5DMBVTFʜ 2*TA'0361%"5&ATBGF  "4BGFCFDBVTF*XSPUFUIJTXJUIJNBHJOH.Z42-`T NJOE
  16. UPDATE `queue`
 JOIN (
 SELECT id
 FROM `queue` FORCE INDEX

    (`index_queue_on_timeout`)
 WHERE 1300000000<timeout AND timeout<=UNIX_TIMESTAMP()
 ORDER BY timeout ASC
 LIMIT :max_acquire FOR UPDATE) AS t1 USING(id)
 SET timeout=:next_timeout, owner=CONNECTION_ID() SELECT id, timeout, data, created_at, resource
 FROM `#{@table}`
 WHERE timeout = ? AND owner = CONNECTION_ID() "DRVJSJOHUBTLTXJUIPVUUBCMFMPDLJOH 4FMFDU.BSL ˣ .BSL4FMFDU #ZUIJTTUSBUFHZ JUBWPJETOFUXPSL MBUFODZJTTVF.PSFPWFSJUDBO BDIJFWFGBTUFSUBTLBDRVJSJOH BCPVUY