कार्य प्रवाह के प्रसंस्करण को व्यवस्थित करने के लिए कतारों का उपयोग किया जाता है। वे कलाकारों के बीच कार्यों के संचय और वितरण के लिए आवश्यक हैं। प्रसंस्करण कार्यों के लिए कतारें अतिरिक्त आवश्यकताएं भी प्रदान कर सकती हैं: वितरण गारंटी, एकमुश्त गारंटी, प्राथमिकता, आदि।

एक नियम के रूप में, तैयार-किए गए संदेश कतारबद्ध सिस्टम का उपयोग किया जाता है (एमक्यू - संदेश कतार), लेकिन कभी-कभी आपको एक तदर्थ कतार या कुछ विशेष एक (उदाहरण के लिए, प्राथमिकता कतार और अपवादों के कारण संसाधित नहीं होने वाले कार्यों को फिर से शुरू करने में देरी) को व्यवस्थित करने की आवश्यकता होती है। ऐसी कतारों के निर्माण पर नीचे चर्चा की जाएगी।

प्रयोज्यता सीमाएँ

प्रस्तावित समाधान समान कार्यों के प्रवाह को संभालने के लिए डिज़ाइन किए गए हैं। वे शिथिल युग्मित प्रणाली और घटकों के बीच पब / उप या संदेश के आयोजन के लिए उपयुक्त नहीं हैं।

एक रिलेशनल डेटाबेस के शीर्ष पर एक कतार छोटे और मध्यम भार (प्रति दिन हजारों कार्यों, दसियों से सैकड़ों कलाकारों) के लिए अच्छी तरह से काम करती है, लेकिन बड़े धागे के लिए एक विशेष समाधान का उपयोग करना बेहतर होता है।

पाँच शब्दों में विधि का सार

select ... for update skip locked 

आधार रेखा

सादगी के लिए, इसके बाद केवल अनन्य कार्य पहचानकर्ताओं को तालिका में संग्रहीत किया जाएगा। किसी तरह का पेलोड जोड़ना मुश्किल नहीं होना चाहिए।

सबसे सरल कतार के लिए तालिका में कार्य और इसकी स्थिति है:

 create table task ( id bigint not null primary key, status integer not null default 0 -- 0 - , 1 -  , 2 -  ); create index task__status__idx on task (status); 

कार्य जोड़ना:

 insert into task (id) values ($1) on conflict (id) do nothing; 

निम्नलिखित कार्य प्राप्त करना:

 with next_task as ( select id from task where status = 0 limit 1 for update skip locked ) update task set status = 1 from next_task where task.id = next_task.id returning task.id; 

कार्य पूर्णता:

 update task set status = 2 where id = $1; 

प्राथमिकता कतार

सरल मामले में, कार्य id इसकी प्राथमिकता है। केवल अगले कार्य के लिए अनुरोध को बदल दिया जाता है - प्रसंस्करण कार्यों के आवश्यक आदेश के साथ order by id क्रमबद्ध स्थिति को जोड़ा जाता है। आपको (status, id) द्वारा एक समग्र सूचकांक बनाने की भी आवश्यकता है।

या, प्राथमिकता के लिए, एक अलग कॉलम जोड़ा जाता है:

 create table task ( id bigint not null primary key, priority integer not null, status integer not null default 0 -- 0 - , 1 -  , 2 -  ); create index task__status__priority__idx on task (status, priority); 

 insert into task (id, priority) values ($1, $2) on conflict (id) do nothing; 

 with next_task as ( select id from task where status = 0 order by priority limit 1 for update skip locked ) update task set status = 1 from next_task where task.id = next_task.id returning task.id; 

हाइलाइट किया गया कॉलम आपको फ्लाई पर कार्य की प्राथमिकता को बदलने की अनुमति देता है।

"गिरे" कार्यों की पुनरावृत्ति के साथ कतार

कार्य के निष्पादन के दौरान कोई त्रुटि या अपवाद हो सकता है। ऐसे मामलों में, कार्य को फिर से कतारबद्ध करना होगा। कभी-कभी कुछ समय के लिए इसके दोहराया निष्पादन के समय को स्थगित करना अभी भी आवश्यक है, उदाहरण के लिए, यदि अपवाद तीसरे पक्ष की सेवा की अस्थायी अनुपलब्धता के कारण है।

 create table task ( id bigint not null primary key, status integer not null default 0, -- 0 - , 1 -  , 2 - , 3 - , 4 -   (  ) attempt integer not null default 0, delayed_to timestamp null, error_text text null ); create index task__status__delayed_to__idx on task (status, delayed_to); 

जैसा कि आप देख सकते हैं, स्थितियों की सूची का विस्तार हुआ है और नए कॉलम जोड़े गए हैं:

• attempt - attempt संख्या; पुन: प्रयास करने की आवश्यकता के बारे में निर्णय लेने की आवश्यकता है (प्रयासों की संख्या को सीमित करें) और पुनः प्रयास करने से पहले एक देरी का चयन करने के लिए (उदाहरण के लिए, प्रत्येक बाद के प्रयास में 10 * attempt मिनट तक देरी हो रही है);
• delayed_to - कार्य को पूरा करने के लिए अगले प्रयास का समय;
• error_text - त्रुटि पाठ।

त्रुटि पाठ को त्रुटि प्रकार के समूह के लिए आवश्यक है।

एक उदाहरण है। निगरानी प्रणाली रिपोर्ट करती है कि स्थिति "त्रुटि" के साथ हजारों कार्य कतार में जमा हो गए हैं। हम अनुरोध पूरा करते हैं:

 select error_text, count(*) from task where status = 3 group by 1 order by 2 desc; 

विवरण के लिए, कलाकारों के लॉग पर जाएं। उस स्थिति को ठीक करें जो त्रुटि का कारण बनी (यदि संभव हो)। यदि आवश्यक हो, तो हम स्थिति को 0 पर सेट करके या अगले प्रयास के समय को शिफ्ट करके कार्यों को पुनः आरंभ करते हैं।

 with next_task as ( select id from task where status = 0 limit 1 for update skip locked ) update task set status = 1, attempt = attempt + 1, delayed_to = null, error_text = null from next_task where task.id = next_task.id returning task.id; 

 with next_task as ( select id from task where status = 3 and delayed_to < localtimestamp limit 1 for update skip locked ) update task set status = 1, attempt = attempt + 1, delayed_to = null, error_text = null from next_task where task.id = next_task.id returning task.id; 

 update task set status = 2, delayed_to = null, error_text = null where id = $1; 

 update task set status = 3, delayed_to = localtimestamp + make_interval(mins => 5 * attempt), error_text = $2 where id = $1; 

 update task set status = 4, delayed_to = null, error_text = $2 where id = $1; 

अगले कार्य के लिए अनुरोध को दो में विभाजित किया गया है ताकि DBMS प्राथमिकता कतार के लिए एक प्रभावी क्वेरी योजना बना सके। के साथ चयन की स्थिति or क्रमबद्धता के साथ बहुत गलत हो सकती order by ।

मेट्रिक्स संग्रह

निम्नलिखित विशेषताएं जोड़ें:

• कार्य निर्माण समय;
• कार्य परिवर्तन समय;
• कार्य का प्रारंभ और समाप्ति समय।

 create table task ( id bigint not null primary key, status integer not null default 0, -- 0 - , 1 -  , 2 - , 3 - , 4 -   (  ) attempt integer not null default 0, begin_time timestamp null, end_time timestamp null, delayed_to timestamp null, error_text text null, created timestamp not null default localtimestamp, updated timestamp not null default localtimestamp ); create index task__status__delayed_to__idx on task (status, delayed_to); create index task__updated__idx on task (updated); 

हम सभी प्रश्नों में जोड़े गए कॉलम पर विचार करते हैं।

 with next_task as ( select id from task where status = 0 limit 1 for update skip locked ) update task set status = 1, attempt = attempt + 1, begin_time = localtimestamp, end_time = null, delayed_to = null, error_text = null, updated = localtimestamp from next_task where task.id = next_task.id returning task.id; 

 with next_task as ( select id from task where status = 3 and delayed_to < localtimestamp limit 1 for update skip locked ) update task set status = 1, attempt = attempt + 1, begin_time = localtimestamp, end_time = null, delayed_to = null, error_text = null, updated = localtimestamp from next_task where task.id = next_task.id returning task.id; 

 update task set status = 2, end_time = localtimestamp, delayed_to = null, error_text = null, updated = localtimestamp where id = $1; 

 update task set status = 3, end_time = localtimestamp, delayed_to = localtimestamp + make_interval(mins => 5 * attempt), error_text = $2, updated = localtimestamp where id = $1; 

 update task set status = 4, end_time = localtimestamp, delayed_to = null, error_text = $2, updated = localtimestamp where id = $1; 

इसकी आवश्यकता क्यों पड़ सकती है, इसके उदाहरण

झूलने वाले कार्यों को खोजें और पुनः आरंभ करें:

 update task set status = 3, end_time = localtimestamp, delayed_to = localtimestamp, error_text = 'hanged', updated = localtimestamp where status = 1 and updated < localtimestamp - interval '1 hour'; 

पुराने कार्यों को हटाना:

 delete from task where updated < localtimestamp - interval '30 days'; 

कार्य पूरा करने के आँकड़े:

 select date_trunc('hour', end_time), count(*), sum(end_time - begin_time), avg(end_time - begin_time) from task where status = 2 and end_time >= '2019-12-16' group by 1 order by 1; 

पहले से पूर्ण किए गए कार्यों को पुनः आरंभ करें

उदाहरण के लिए, एक दस्तावेज़ अपडेट किया गया है, आपको पूर्ण-पाठ खोज के लिए इसे फिर से लिखना होगा।

 create table task ( id bigint not null primary key, task_updated_at timestamp not null default localtimstamp, status integer not null default 0, -- 0 - , 1 -  , 2 - , 3 - , 4 -   (  ) begin_time timestamp null, end_time timestamp null, delayed_to timestamp null, error_text text null, created timestamp not null default localtimestamp, updated timestamp not null default localtimestamp ); 

यहां, टास्क अपडेट समय के लिए task_updated_at कॉलम जोड़ा जाता है, लेकिन created फ़ील्ड का उपयोग किया जा सकता है।

किसी कार्य को जोड़ना या अपडेट करना (पुनः आरंभ करना):

 insert into task (id, task_updated_at) values ($1, $2) on conflict (id) do update set task_updated_at = excluded.task_updated_at, status = case when status = 1 then 1 else 0 end, delayed_to = null, error_text = null, updated = localtimestamp where task_updated_at < excluded.task_updated_at; 

यहाँ क्या चल रहा है यदि यह अभी पूरा नहीं हो रहा है तो एक कार्य "नया" हो जाता है।

कार्य को पूरा करने का अनुरोध भी जांच करेगा कि क्या निष्पादन के दौरान इसे बदल दिया गया था।

अगले कार्य के लिए अनुरोध मेट्रिक्स इकट्ठा करने के लिए कतार में जैसे हैं।

कार्य का सफल समापन:

 update task set status = case when begin_time >= updated then 2 else 0 end, end_time = localtimestamp, delayed_to = null, error_text = null, updated = localtimestamp where id = $1; 

त्रुटि के साथ कार्य पूरा होना: कार्य के आधार पर। आप पुनरारंभ करने में बिना शर्त देरी कर सकते हैं, आप अपडेट करते समय स्थिति को "नया" सेट कर सकते हैं।

पाइपलाइन

कार्य कई चरणों से गुजरता है। आप प्रत्येक चरण के लिए एक अलग कतार बना सकते हैं। या आप तालिका में संबंधित कॉलम जोड़ सकते हैं।

मूल कतार पर आधारित एक उदाहरण ताकि कोड को अव्यवस्थित न किया जा सके। पहले से वर्णित सभी संशोधनों को बिना किसी समस्या के इस कतार में लागू किया जा सकता है।

 create table task ( id bigint not null primary key, stage integer not null default 0, status integer not null default 0 ); create index task__stage__status__idx on task (stage, status); 

दिए गए चरण में निम्नलिखित कार्य करना:

 with next_task as ( select id from task where stage = $1 and status = 0 limit 1 for update skip locked ) update task set status = 1 from next_task where task.id = next_task.id returning task.id; 

संकेतित चरण में संक्रमण के साथ कार्य पूरा करना:

 update task set stage = $2, status = 2 where id = $1; 

या क्रम में अगले चरण में संक्रमण:

 update task set stage = stage + 1, status = 2 where id = $1; 

अनुसूचित कार्य

यह दोहराने की कतार का एक रूपांतर है।

प्रत्येक कार्य का अपना कार्यक्रम हो सकता है (सरलतम संस्करण में, लॉन्च की आवृत्ति)।

 create table task ( id bigint not null primary key, period integer not null, --     status integer not null default 0, -- 0 - , 1 -   next_run_time timestamp not null default localtimestamp ); create index task__status__next_run_time__idx on task (status, next_run_time); 

कार्य जोड़ना:

 insert into task (id, period, next_run_time) values ($1, $2, $3); 

निम्नलिखित कार्य प्राप्त करना:

 with next_task as ( select id from task where status = 0 and next_run_time <= localtimestamp limit 1 for update skip locked ) update task set status = 1 from next_task where task.id = next_task.id returning task.id; 

कार्य को पूरा करना और अगले रन की योजना बनाना:

 update task set status = 0, next_run_time = next_run_time + make_interval(secs => period) where id = $1 

एक निष्कर्ष के बजाय

RDBMS टूल का उपयोग करके एक विशेष कार्य कतार बनाने में कुछ भी जटिल नहीं है।

"स्व-निर्मित" कतार जवाब देगी यहां तक ​​कि सबसे जंगली वस्तुतः किसी भी व्यवसाय / डोमेन की आवश्यकता।

खैर, हमें यह नहीं भूलना चाहिए कि किसी भी अन्य डेटाबेस की तरह, कतार में सर्वर, क्वेरी और इंडेक्स की विचारशील ट्यूनिंग की आवश्यकता होती है।