Python sqlite3 i współbieżność

Mam program w języku Python, który korzysta z modułu „wątkowania”. Co sekundę mój program uruchamia nowy wątek, który pobiera dane z sieci i zapisuje je na moim dysku twardym. Chciałbym używać sqlite3 do przechowywania tych wyników, ale nie mogę zmusić go do działania. Wydaje się, że problem dotyczy następującej linii:

conn = sqlite3.connect("mydatabase.db")

• Jeśli umieszczę ten wiersz kodu w każdym wątku, otrzymam OperationalError informujący mnie, że plik bazy danych jest zablokowany. Myślę, że oznacza to, że inny wątek ma otwarty plik mydatabase.db za pośrednictwem połączenia sqlite3 i zablokował go.
• Jeśli umieszczę ten wiersz kodu w programie głównym i przekażę obiekt połączenia (conn) do każdego wątku, otrzymuję błąd ProgrammingError, który mówi, że obiekty SQLite utworzone w wątku mogą być używane tylko w tym samym wątku.

Wcześniej zapisywałem wszystkie wyniki w plikach CSV i nie miałem żadnego z tych problemów z blokowaniem plików. Miejmy nadzieję, że będzie to możliwe dzięki sqlite. Jakieś pomysły?

Możesz użyć wzorca konsument-producent. Na przykład możesz utworzyć kolejkę współdzieloną między wątkami. Pierwszy wątek, który pobiera dane z sieci, umieszcza te dane w kolejce udostępnionej. Inny wątek, który jest właścicielem połączenia z bazą danych, usuwa dane z kolejki z kolejki i przekazuje je do bazy danych.

Wbrew powszechnemu przekonaniu, nowsze wersje sqlite3 zrobić support dostęp z wielu wątków.

Można to włączyć za pomocą opcjonalnego argumentu słowa kluczowego check_same_thread:

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

Następujące znalezione na mail.python.org.pipermail.1239789

znalazłem rozwiązanie. Nie wiem, dlaczego w dokumentacji Pythona nie ma ani słowa o tej opcji. Musimy więc dodać nowy argument słowa kluczowego do funkcji połączenia i będziemy mogli stworzyć z niego kursory w innym wątku. Więc użyj:

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

u mnie działa idealnie. Oczywiście od teraz muszę dbać o bezpieczny wielowątkowy dostęp do bazy danych. W każdym razie dzięki wszystkim za próby pomocy.

Przejdź do przetwarzania wieloprocesowego . Jest znacznie lepszy, dobrze skaluje się, może wykraczać poza użycie wielu rdzeni przy użyciu wielu procesorów, a interfejs jest taki sam, jak w przypadku korzystania z modułu wątkowości w Pythonie.

Lub, jak zasugerował Ali, po prostu użyj mechanizmu buforowania wątków SQLAlchemy . Zajmie się wszystkim automatycznie i ma wiele dodatkowych funkcji, wystarczy zacytować niektóre z nich:

1. SQLAlchemy zawiera dialekty dla SQLite, Postgres, MySQL, Oracle, MS-SQL, Firebird, MaxDB, MS Access, Sybase i Informix; IBM wydał również sterownik DB2. Nie musisz więc przepisywać aplikacji, jeśli zdecydujesz się odejść od SQLite.
2. System Unit Of Work, centralna część Object Relational Mapper (ORM) SQLAlchemy, organizuje oczekujące operacje tworzenia / wstawiania / aktualizowania / usuwania w kolejkach i opróżnia je wszystkie w jednej partii. Aby to osiągnąć, wykonuje topologiczne „sortowanie według zależności” wszystkich zmodyfikowanych elementów w kolejce, tak aby uwzględniać ograniczenia klucza obcego, i grupuje nadmiarowe instrukcje razem, gdzie czasami można je grupować jeszcze bardziej. Zapewnia to maksymalną wydajność i bezpieczeństwo transakcji oraz minimalizuje ryzyko zakleszczenia.

W ogóle nie powinieneś używać do tego wątków. To banalne zadanie dla twisted które i tak prawdopodobnie zabrałoby cię znacznie dalej.

Użyj tylko jednego wątku i pozwól, aby zakończenie żądania wyzwoliło zdarzenie do wykonania zapisu.

twisted zajmie się harmonogramem, oddzwonieniami itp. za Ciebie. Przekaże ci cały wynik jako ciąg znaków lub możesz go uruchomić przez procesor strumieniowy (mam API Twittera i Friendfeed API, które odpalają zdarzenia do wywołujących, ponieważ wyniki są nadal pobierane).

W zależności od tego, co robisz ze swoimi danymi, możesz po prostu zrzucić pełny wynik do sqlite, gdy jest gotowy, ugotować go i zrzucić lub ugotować podczas odczytu i zrzucić na końcu.

Mam bardzo prostą aplikację, która robi coś zbliżonego do tego, czego oczekujesz na githubie. Nazywam to pfetch (pobieranie równoległe). Przechwytuje różne strony zgodnie z harmonogramem, przesyła wyniki do pliku i opcjonalnie uruchamia skrypt po pomyślnym zakończeniu każdej z nich. Wykonuje również kilka wymyślnych rzeczy, takich jak warunkowe GET, ale nadal może być dobrą bazą do wszystkiego, co robisz.

Lub jeśli jesteś leniwy, jak ja, możesz użyć SQLAlchemy . Zajmie się obsługą wątków za Ciebie ( używając lokalnego wątku i niektórych pul połączeń ), a sposób, w jaki to robi, jest nawet konfigurowalny .

Dla dodatkowego bonusu, jeśli / kiedy zdasz sobie sprawę / zdecydujesz, że użycie Sqlite dla dowolnej aplikacji współbieżnej będzie katastrofą, nie będziesz musiał zmieniać swojego kodu, aby używać MySQL, Postgres lub cokolwiek innego. Możesz po prostu przełączyć.

Musisz użyć session.close()po każdej transakcji do bazy danych, aby użyć tego samego kursora w tym samym wątku, nie używając tego samego kursora w wielu wątkach, które powodują ten błąd.

Użyj threading.Lock ()

Podoba mi się odpowiedź Evgeny'ego - kolejki są generalnie najlepszym sposobem implementacji komunikacji między wątkami. Aby uzyskać kompletność, oto kilka innych opcji:

• Zamknij połączenie DB, gdy odrodzone wątki zakończą jego używanie. To naprawiłoby twoje OperationalError, ale otwieranie i zamykanie połączeń w ten sposób jest generalnie nie-nie, ze względu na narzut wydajności.
• Nie używaj wątków podrzędnych. Jeśli zadanie wykonywane raz na sekundę jest dość lekkie, możesz uciec od pobierania i zapisywania, a następnie spania do odpowiedniego momentu. Jest to niepożądane, ponieważ operacje pobierania i przechowywania mogą zająć> 1 s, a użytkownik traci korzyści wynikające z multipleksowanych zasobów, które mają przy podejściu wielowątkowym.

Musisz zaprojektować współbieżność dla swojego programu. SQLite ma wyraźne ograniczenia i musisz ich przestrzegać, zobacz FAQ (także poniższe pytanie).

Scrapy wydaje się być potencjalną odpowiedzią na moje pytanie. Jego strona główna opisuje dokładnie moje zadanie. (Chociaż nie jestem pewien, jak stabilny jest kod).

Chciałbym rzucić okiem na moduł y_serial Python dla trwałości danych: http://yserial.sourceforge.net

który obsługuje problemy z zakleszczeniami otaczającymi pojedynczą bazę danych SQLite. Jeśli zapotrzebowanie na współbieżność stanie się duże, można łatwo skonfigurować klasę Farm wielu baz danych, aby rozproszyć obciążenie w czasie stochastycznym.

Mam nadzieję, że to pomoże Twojemu projektowi ... powinien być wystarczająco prosty do wdrożenia w 10 minut.

Nie mogłem znaleźć żadnych punktów odniesienia w żadnej z powyższych odpowiedzi, więc napisałem test, aby wszystko porównać.

Wypróbowałem 3 podejścia

1. Czytanie i pisanie sekwencyjnie z bazy danych SQLite
2. Używanie ThreadPoolExecutor do odczytu / zapisu
3. Używanie ProcessPoolExecutor do odczytu / zapisu

Wyniki i wnioski z testu porównawczego są następujące

1. Odczyty sekwencyjne / zapisy sekwencyjne działają najlepiej
2. Jeśli musisz przetwarzać równolegle, użyj ProcessPoolExecutor do odczytu równoległego
3. Nie wykonuj żadnych zapisów ani za pomocą ThreadPoolExecutor, ani ProcessPoolExecutor, ponieważ napotkasz błędy zablokowane w bazie danych i będziesz musiał ponownie spróbować wstawić fragment

Możesz znaleźć kod i kompletne rozwiązanie dla testów porównawczych w mojej odpowiedzi SO TUTAJ Mam nadzieję, że to pomoże!

Najbardziej prawdopodobnym powodem otrzymywania błędów z zablokowanymi bazami danych jest konieczność wydania

conn.commit()

po zakończeniu operacji na bazie danych. Jeśli tego nie zrobisz, twoja baza danych zostanie zablokowana do zapisu i taka pozostanie. Pozostałe wątki, które czekają na zapis, po pewnym czasie wygaśnie (domyślnie jest to 5 sekund, szczegóły na http://docs.python.org/2/library/sqlite3.html#sqlite3.connect ) .

Przykład poprawnego i równoczesnego wstawiania wyglądałby tak:

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()

Jeśli lubisz SQLite lub masz inne narzędzia, które współpracują z bazami danych SQLite, lub chcesz zamienić pliki CSV na pliki SQLite db lub musisz zrobić coś rzadkiego, jak międzyplatformowe IPC, to SQLite jest świetnym narzędziem i bardzo pasuje do tego celu. Nie daj się zmuszać do korzystania z innego rozwiązania, jeśli nie wydaje się to właściwe!