Как заблокировать строку для ограничения скорости нескольких рабочих?

Question

У меня есть несколько воркеров, распределенных по нескольким узлам, которые очищают HTML. Мне нужно указать ограничение скорости в системе, чтобы ни один домен не получал более 1 запроса каждые 5 секунд.

Каждый рабочий имеет доступ к общей базе данных (PostgreSQL), поэтому я создал таблицу с двумя столбцами:

ключ домена, дата последнего сканирования

В рабочем коде я хочу проверить дату последнего сканирования перед отправкой запроса. Проблема в том, что тысячи рабочих могут получить один и тот же домен почти в один и тот же момент, если задачи распределены циклически, поэтому, если они все читают сразу, они не увидят недавнего сканирования и все запросы будут отправлены. Поэтому мне нужен способ заблокировать поле, чтобы первый проверяемый работник включил блокировку, выполнил сканирование, а затем обновил дату сканирования и снял блокировку. Затем все другие рабочие могут проверить, существует ли блокировка в строке, и отклонить задачу, чтобы она была перенесена в расписание.

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

Какой тип замка мне нужен? Есть ли какие-нибудь хорошие примеры, показывающие этот тип блокировки?

Если я просто заключу каждый процесс в транзакцию, это сработает?

Answer 1

Ваш основной код будет блоком:

begin;
  set transaction isolation level read committed; -- should already be the default
  select domain_key
    from your_table
   where last_scan_date < now() - interval '5 seconds'
     for update skip locked
   limit 1;
-- Do your stuff here, and issue a rollback if it fails
  update your_table
     set last_scan_date = <value goes here>
   where domain_key = <value goes here>;
commit;

Я ожидаю, что он будет использоваться на языке хоста. Следующий пример фрагмента воркера находится в python:

conn = psycopg2.connect('<db connect parameters>')
conn.autocommit = false
c = conn.cursor()
c.execute("set transaction isolation level read committed;") 
c.execute("""
   select domain_key
     from your_table
    where last_scan_date < now() - interval '5 seconds'
    order by last_scan_date  
      for update skip locked
    limit 1
""")
domain_key = c.fetchone()[0]
if domain_key:
    result = process_url(domain_key)  # <-- This is your scraping routine
    if result == 'Ok':
        c.execute("""
          update your_table
             set last_scan_date = now() 
           where domain_key = %s
        """, (domain_key,))
        conn.commit()
    else:
        conn.rollback()