Как правильно реализовать мультипоточность?

Question

[stayHARD]

Здравствуйте.
Есть небольшой парсер сайтов, необходимо заставить его работать быстрее.
Структура:

# блок с импортами
import ....

# блок с подключением к БД (оттуда нужно забрать урлы по которым пройтись)
conn = sqlite3.connect('db.sqlite3')
curs = conn.cursor()
select = curs.execute("SELECT url from test;")

# блок с основной функцией, которая забирает данные с сайта
def scrape(link):
    ....
    # блок записи в БД, используя подключение выше
    return "Success"

# блок, который вызывает функцию scrape для каждого элемента, что был взят из БД
for link in select.fetchall():
    scrape(link)

Пытался добавить мультипоточность(к сожалению безуспешно), используя модуль - threading
И данный код:

thread_list = []

t = threading.Thread(target=scrape, args=(link,))
thread_list.append(t)

for thread in thread_list:
	thread.start()

for thread in thread_list:
    thread.join()

Comments

[stayHARD]

pygame:
OperationalError: cannot start a transaction within a transaction
ProgrammingError: Recursive use of cursors not allowed.
OperationalError: not an error
OperationalError: database is locked

Answer 1

[Олег]

Предлагаю вам свой вариант, через очередь задач:

from Queue import Queue
from threading import Thread

def scrape(link):
    conn = sqlite3.connect('db.sqlite3')
    curs = conn.cursor()
    # insert link, close connection

class Worker(Thread):

    def __init__(self, tasks):
        super(Worker, self).__init__()
        self.tasks = tasks
        self.daemon = True

    def run(self):
        while True:
            link = self.tasks.get()
            try:
                scrape(link)
            finally:
                self.tasks.task_done()


if __name__ == '__main__':
    # максимальное количество одновременных потоков
    capacity = 0  # infinite
    queue = Queue(capacity)

    workers = 3
    for _ in range(workers):
        Worker(queue).start()

    for link in select.fetchall():
        queue.put(link)

    queue.join()
    print 'Done'

Comments

[stayHARD]

Получаю набор из вот таких ошибок:
OperationalError: cannot start a transaction within a transaction
ProgrammingError: Recursive use of cursors not allowed.
OperationalError: not an error

[Олег]

Перенесите создание нового коннекшена к sqlite и курсора в функцию scrape, чтобы каждый раз при ее вызове все создавалось по новой, скорее всего вы используете в ней уже созданный глобальный объект.

[angru]

Олег , а зачем вам очередь, если вы все равно создаете по воркеру на каждый линк?

[Олег]

angru: ну это был быстрый пример) я его немного изменил, чтобы воркеров было другое количество и код приобрел больше смысла, спасибо)

Answer 2

[lega]

Для парсера сайтов лучше использовать асинхронный фреймворк, пример www.py-my.ru/post/4f278211bbddbd0322000000

Comments

[stayHARD]

Это хорошо, но парсер уже написан. Необходимо только добавить мультипотоки.
Возможно проблема в sqlite3, если использовать что-то другое, проблем не будет?

Answer 3

[angru]

Как посоветовали выше - используйте какой-нибудь асинхронный фреймворк(мне в свое время понравился gevent). Вам и так и так надо вносить изменения, будь то потоки или асинхронность, а асинхронность на мой взгляд лучше подходит в этом случае, т.к у вас в основном IO-операции, к тому же весь питоновский код все равно выполняется на одном ядре(гуглить GIL) да и SQLite не может потоки, так что не уверен, что потоки дадут вам какие-то преимущества.

Comments

[angru]

from gevent import monkey; monkey.patch_all()

from gevent import spawn, sleep, joinall
from gevent.queue import Queue
from gevent.pool import Pool


def scrape(url):
    return data


def fetch(url, queue):
    queue.put(scrape(url))


def start(urls, queue):
    pool = Pool(size=20)
    joinall([pool.spawn(fetch, url, queue) for url in urls])


def main_loop():
    urls = get_urls_from_db()
    queue = Queue(maxsize=20)
    tasks = spawn(start, urls, queue)
    write_tasks = []

    while not g.ready() or not queue.empty():
        result = queue.get()

        write_tasks.append(spawn(write_into_db, result))
        sleep(0)
    
     joinall(write_tasks)

Answer 4

[sim3x]

sqlite3 файловая однопоточная СУБД. Не стоит ее использовать в многопоточном режиме

Для парсинга используй scrapy

Если сильно хочется самому все написать, то stackoverflow.com/a/6125654