Параллельная синхронизация, избегая ключевых нарушений

Question

У меня есть несколько баз данных, которые синхронизируются с использованием таблицы журнала:

source bigint,
tick serial,
data text

primary key (source, tick)

Таким образом, наивный подход к синхронизации, например, заключается в том, что мой второй источник синхронизирует данные из первого источника:

insert into log_table (source, tick, data)
select source, tick, data
from other_db.log_table
where source = 1
and tick > (select max(tick) from log_table where source = 1)

Я упростил пример.На практике это использует dblink для запроса другой базы данных, но я не думаю, что это имеет отношение к этой проблеме.Проблема, с которой я столкнулся, заключается в том, что, если у меня несколько потоков, синхронизирующихся, я получаю нарушения ключа.

Мой текущий способ - обнаружить нарушение ключа и просто повторить попытку.Вполне вероятно, что эта повторная попытка ничего не обработает, так как другой поток выполнит эту работу.Мне интересно, может быть, есть более эффективный подход?

Answer 1

Если вы используете postgres 9.6 или выше, вы можете попробовать выражение upsert on duplicate key do nothing (или on duplicate key do update зависит от того, что вы хотите сделать)

https://www.postgresql.org/docs/9.6/static/sql-insert.html

с вашим запросом

insert into log_table (source, tick, data)
select source, tick, data
    from other_db.log_table
    where source = 1
        and tick > (select max(tick) from log_table where source = 1)
on duplicate key do nothing

Или вы можете попробовать использовать хранимую процедуру, в которой вы сможете использовать блокировки на уровне строк https://www.postgresql.org/docs/9.6/static/explicit-locking.html

Например, вы можете создать функцию, которая будет выглядеть примерно так

create or replace function tick_update() returns void
language plpgsql as
$f$
declare
  max_tick integer;
begin
    select max(tick) into max_tick from log_table where source = 1;
    for rec in 
        select source, tick, data
        from other_db.log_table
        where source = 1
            and tick > max_tick
        for update
    loop
    insert into log_table (source, tick, data) values (rec.source, rec.tick, rec.data)
    end loop;
end
$f$