PostgreSQL - implementazione di una coda affidabile

Question

Sto cercando di implementare una coda affidabile con più writer e più lettori utilizzando il database Postgres. Come evitare le righe mancanti quando un lettore di code esegue la scansione di una tabella, quindi le transazioni in corso vengono confermate dopo la lettura.

Abbiamo un lettore che seleziona le righe in batch utilizzando un tempo di "checkpoint", in cui ogni batch riceve le righe dopo l'ultimo timestamp nel batch precedente e mancano le righe. (Motivo: il valore del timestamp si basa sull'ora in cui si verifica INSERT (00.00.00). A carichi pesanti, se la transazione richiede più tempo, viene inserita diciamo 10 secondi dopo (00.00.10), il lettore salterà questa riga (row1) se legge durante quei 10 secondi e trova una riga che ha avuto il suo tempo INSERT in un secondo momento (00.00.05) rispetto alla riga 1. La descrizione completa del problema è simile a quella scritta in questo blog http://blog.thefourthparty.com/stopping-time-in-postgresql/)

Related domanda prima per il contesto: Postgres LISTEN/NOTIFY - low latency, realtime?

Aggiornamento: avevo aggiornato la questione di dover singolo lettore a più lettori. L'ordine in cui il lettore legge è importante.

Answer 1

Considerando più lettori, è necessario avere il controllo su quali record sono già stati ricevuti da ciascun lettore.

Inoltre, è stato detto che l'ordine è una condizione per inviare record anche a un lettore. Quindi, se qualche ulteriore transazione è stata commessa prima di una precedente, dobbiamo "fermarci" e inviare nuovamente i record quando si è impegnato, per mantenere l'ordine dei record inviati al lettore.

Detto questo, verificare l'attuazione:

-- 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;