एकाधिक पाठकों को ध्यान में रखते हुए, यह नियंत्रित करना आवश्यक है कि प्रत्येक पाठक को पहले से ही कौन से रिकॉर्ड प्राप्त हुए हों।
इसके अलावा, यह कहा गया है कि ऑर्डर एक पाठक को रिकॉर्ड भेजने की शर्त है। इसलिए, यदि कुछ और लेनदेन पहले के पहले किए गए थे, तो हमें पाठक को भेजे गए रिकॉर्ड के आदेश को बनाए रखने के लिए, "रोक" देना होगा और जब उसने प्रतिबद्ध किया था, तो रिकॉर्ड फिर से भेजना होगा।
जिसके अनुसार, कार्यान्वयन की जाँच करें:
-- lets create our queue table
drop table if exists queue_records cascade;
create table if not exists queue_records
(
cod serial primary key,
date_posted timestamp default timeofday()::timestamp,
message text
);
-- lets create a table to save "checkpoints" per reader_id
drop table if exists queue_reader_checkpoint cascade;
create table if not exists queue_reader_checkpoint
(
reader_id text primary key,
last_checkpoint numeric
);
CREATE OR REPLACE FUNCTION get_queue_records(pREADER_ID text)
RETURNS SETOF queue_records AS
$BODY$
DECLARE
vLAST_CHECKPOINT numeric;
vCHECKPOINT_EXISTS integer;
vRECORD queue_records%rowtype;
BEGIN
-- let's get the last record sent to the reader
SELECT last_checkpoint
INTO vLAST_CHECKPOINT
FROM queue_reader_checkpoint
WHERE reader_id = pREADER_ID;
-- if vLAST_CHECKPOINT is null (this is the very first time of reader_id),
-- sets it to the last cod from queue. It means that reader will get records from now on.
if (vLAST_CHECKPOINT is null) then
-- sets the flag indicating the reader does not have any checkpoint recorded
vCHECKPOINT_EXISTS = 0;
-- gets the very last commited record
SELECT MAX(cod)
INTO vLAST_CHECKPOINT
FROM queue_records;
else
-- sets the flag indicating the reader already have a checkpoint recorded
vCHECKPOINT_EXISTS = 1;
end if;
-- now let's get the records from the queue one-by-one
FOR vRECORD IN
SELECT *
FROM queue_records
WHERE COD > vLAST_CHECKPOINT
ORDER BY COD
LOOP
-- if next record IS EQUALS to (vLAST_CHECKPOINT+1), the record is in the expected order
if (vRECORD.COD = (vLAST_CHECKPOINT+1)) then
-- let's save the last record read
vLAST_CHECKPOINT = vRECORD.COD;
-- and return it
RETURN NEXT vRECORD;
-- but, if it is not, then is out of order
else
-- the reason is some transaction did not commit yet, but there's another further transaction that alread did.
-- so we must stop sending records to the reader. And probably next time he calls, the transaction will have committed already;
exit;
end if;
END LOOP;
-- now we have to persist the last record read to be retrieved on next call
if (vCHECKPOINT_EXISTS = 0) then
INSERT INTO queue_reader_checkpoint (reader_id, last_checkpoint) values (pREADER_ID, vLAST_CHECKPOINT);
else
UPDATE queue_reader_checkpoint SET last_checkpoint = vLAST_CHECKPOINT where reader_id = pREADER_ID;
end if;
end;
$BODY$ LANGUAGE plpgsql VOLATILE;
http://stackoverflow.com/q/6507475/330315 और http://stackoverflow.com/q/22765054/330315 मदद कर सकता है। आप इसे भी देखना चाहेंगे: http://pgxn.org/dist/pg_message_queue/ –
क्या यह महत्वपूर्ण है कि सभी पंक्तियों को * क्रमशः * संसाधित किया जाए? फिर ऑर्डर कैसे परिभाषित किया गया है * ठीक *? या आप बस गायब पंक्तियों से बचना चाहते हैं? फिर यहां प्रस्तुत किए गए समाधान को काम करना चाहिए: [पोस्टग्रेज़ अपडेट, सीमा 1] (http://dba.stackexchange.com/a/69497/3684) –
क्या वास्तव में यह आवश्यक है कि आप postgresql के साथ ऐसा करें? यह ऐसी आवश्यकता है जो बहुत आसानी से पूर्ण हो सकती है redis – e4c5