Python sqlite3 и параллелизм

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

Мне нравится ответ Евгения: очереди - это вообще лучший способ реализовать межпотоковое взаимодействие. Для полноты картины вот еще несколько вариантов:

• Закройте соединение с БД, когда порожденные потоки закончат его использовать. Это исправит ваш OperationalError, но открывать и закрывать такие соединения, как правило, нельзя из-за накладных расходов на производительность.
• Не используйте дочерние потоки. Если задача, выполняемая один раз в секунду, достаточно легкая, вам может сойти с рук выполнение выборки и сохранения, а затем засыпание до подходящего момента. Это нежелательно, поскольку операции выборки и сохранения могут занимать> 1 секунды, и вы теряете преимущества мультиплексированных ресурсов, которые у вас есть при многопоточном подходе.

Переключитесь на многопроцессорность. Он намного лучше, хорошо масштабируется, может выходить за рамки использования нескольких ядер за счет использования нескольких процессоров, а интерфейс такой же, как при использовании модуля потоковой передачи Python.

Или, как предложил Али, просто используйте Механизм пула потоков SQLAlchemy. Он сделает все за вас автоматически и имеет множество дополнительных функций, просто процитирую некоторые из них:

1. SQLAlchemy включает диалекты для SQLite, Postgres, MySQL, Oracle, MS-SQL, Firebird, MaxDB, MS Access, Sybase и Informix; IBM также выпустила драйвер DB2. Таким образом, вам не нужно переписывать свое приложение, если вы решите отказаться от SQLite.
2. Система Unit Of Work, центральная часть Object Relational Mapper (ORM) SQLAlchemy, организует ожидающие операции создания / вставки / обновления / удаления в очереди и сбрасывает их все в одном пакете. Для этого он выполняет топологическую «сортировку зависимостей» всех измененных элементов в очереди, чтобы соблюдать ограничения внешнего ключа, и группирует избыточные операторы вместе, где они иногда могут быть дополнительно объединены в пакеты. Это обеспечивает максимальную эффективность и безопасность транзакций и сводит к минимуму вероятность возникновения взаимоблокировок.

Или, если вы ленивы, как я, можете использовать SQLAlchemy. Он будет обрабатывать потоки за вас, (с использованием локального потока и некоторого пула соединений), и способ, которым он это делает, даже настраиваемый.

В качестве дополнительного бонуса, если / когда вы поймете / решите, что использование Sqlite для любого параллельного приложения будет катастрофой, вам не придется менять свой код для использования MySQL, Postgres или чего-либо еще. Вы можете просто переключиться.

Вам необходимо спроектировать параллелизм для вашей программы. SQLite имеет четкие ограничения, и вам необходимо соблюдать их, см. Часто задаваемые вопросы (также следующий вопрос).

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

Используйте только один поток, и завершение запроса инициирует событие для выполнения записи.

twisted позаботится о расписании, обратных вызовах и т. д. за вас. Он передаст вам весь результат в виде строки, или вы можете запустить его через потоковый процессор (у меня есть twitter API и Friendfeed API, которые оба запускают события для вызывающих, поскольку результаты все еще загружаются).

В зависимости от того, что вы делаете со своими данными, вы можете просто выгрузить полный результат в sqlite по мере его завершения, приготовить его и выгрузить или приготовить его во время чтения и выгрузить в конце.

У меня очень простое приложение, которое делает что-то похожее на то, что вы хотите от github. Я называю это pfetch (параллельная выборка). Он захватывает различные страницы по расписанию, передает результаты в файл и, при необходимости, запускает сценарий после успешного завершения каждой из них. Он также делает некоторые причудливые вещи, такие как условные GET, но все же может быть хорошей базой для всего, что вы делаете.

Scrapy кажется потенциальным ответом на мой вопрос. Его домашняя страница точно описывает мою задачу. (Хотя я еще не уверен, насколько стабилен код.)

Я бы посмотрел на модуль Python y_serial для сохранения данных: http://yserial.sourceforge.net

который обрабатывает проблемы взаимоблокировки, связанные с одной базой данных SQLite. Если спрос на параллелизм становится большим, можно легко настроить класс Farm для многих баз данных, чтобы распределить нагрузку в течение стохастического времени.

Надеюсь, это поможет вашему проекту ... он должен быть достаточно простым, чтобы реализовать его за 10 минут.

Следующее найдено на mail.python.org.pipermail.1239789

Я нашел решение. Я не знаю, почему в документации Python нет ни слова об этой опции. Итак, мы должны добавить новый аргумент ключевого слова в функцию подключения и мы сможем создавать из него курсоры в другом потоке. Так что используйте:

sqlite.connect(":memory:", check_same_thread = False)

отлично работает для меня. Конечно, с этого момента мне нужно заботиться безопасного многопоточного доступа к БД. В любом случае спасибо всем за попытку помочь.

Вопреки распространенному мнению, более новые версии sqlite3 делать поддерживают доступ из нескольких потоков.

Это можно включить с помощью необязательного ключевого аргумента check_same_thread:

sqlite.connect(":memory:", check_same_thread=False)

Используйте threading.Lock ()

Наиболее вероятная причина, по которой вы получаете ошибки с заблокированными базами данных, заключается в том, что вы должны выпустить

conn.commit()

после завершения операции с базой данных. Если вы этого не сделаете, ваша база данных будет заблокирована от записи и останется такой же. Остальные потоки, ожидающие записи, будут отключены по истечении некоторого времени (по умолчанию установлено 5 секунд, подробности см. В http://docs.python.org/2/library/sqlite3.html#sqlite3.connect).

Пример правильной и одновременной вставки будет следующим:

import threading, sqlite3
class InsertionThread(threading.Thread):

    def __init__(self, number):
        super(InsertionThread, self).__init__()
        self.number = number

    def run(self):
        conn = sqlite3.connect('yourdb.db', timeout=5)
        conn.execute('CREATE TABLE IF NOT EXISTS threadcount (threadnum, count);')
        conn.commit()

        for i in range(1000):
            conn.execute("INSERT INTO threadcount VALUES (?, ?);", (self.number, i))
            conn.commit()

# create as many of these as you wish
# but be careful to set the timeout value appropriately: thread switching in
# python takes some time
for i in range(2):
    t = InsertionThread(i)
    t.start()

Если вам нравится SQLite или у вас есть другие инструменты, которые работают с базами данных SQLite, или вы хотите заменить файлы CSV файлами SQLite db, или вам нужно сделать что-то редкое, например межплатформенный IPC, то SQLite - отличный инструмент и очень подходит для этой цели. Не позволяйте заставлять себя использовать другое решение, если оно вам не подходит!

Я не смог найти никаких тестов ни в одном из приведенных выше ответов, поэтому я написал тест для тестирования всего.

Пробовал 3 подхода

1. Последовательное чтение и запись из базы данных SQLite
2. Использование ThreadPoolExecutor для чтения / записи
3. Использование ProcessPoolExecutor для чтения / записи

Результаты и выводы теста следующие.

1. Последовательное чтение / последовательная запись работают лучше всего
2. Если вы должны обрабатывать параллельно, используйте ProcessPoolExecutor для параллельного чтения.
3. Не выполняйте никаких операций записи ни с помощью ThreadPoolExecutor, ни с помощью ProcessPoolExecutor, так как вы столкнетесь с ошибками блокировки базы данных, и вам придется повторить попытку вставки фрагмента снова.

Вы можете найти код и полное решение для тестов в моем SO-ответе ЗДЕСЬ Надеюсь, это поможет!

Вам необходимо использовать session.close() после каждая транзакция в базе данных, чтобы использовать тот же курсор в том же потоке, не используя тот же курсор в многопоточности, которая вызывает эту ошибку.