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

Question

[astrotrain]

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

import threading
import thread
from threading import Lock
import sys
from random import randint
import time

my_lock = Lock()
class printer(threading.Thread):
    def __init__(self, string):
        threading.Thread.__init__(self)
        self.string = string
    #with my_lock:
    #   print string
    def run(self):
        sys.stdout.write(self.string + '\n')
        sleep_time = randint(1, 10)
        time.sleep(sleep_time)


max_threads = 5
f = open("strings.txt", "r")
data = f.readlines()
data_count = len(data)
print data_count
threads = []
i = 0
flag = True
curr_threads = 0

while(flag == True):


    if(i >= data_count):
        flag = False

    if(flag != False):
        curr_threads = len(threads)
        while(curr_threads < max_threads):
            print i
            data_string = data[i]
            t = printer(data_string) 
            threads.append(t)
            t.daemon = True
            t.start()
            i = i + 1
            if ( i >= data_count):
                break

    for thr in threads:
        t_id = str(thr.ident)
        print "checking " + t_id
        if thr.is_alive():
            t.join()
            print "handling"
            threads.remove(t)

Но, во-первых почему-то засыпает основной поток, а во-вторых я не знаю как правильно обрабатывать работающие потоки. Я думал, что нужно по всем работающим потоком пройтись и если они isAlive то сделать join и удалить его из списка и так далее. Но join блокирует основную программу, поэтому если какой-то поток долго "спит" то это непродуктивно, да и удалить из списка у меня не получается: ValueError: list.remove(x): x not in list. Как правильно разрулить ситуацию?

Answer 1

[angru]

import sys
import time
import Queue
import threading
import random


MAX_THREADS = 5


def start_thread(*args, **kwargs):
    t = threading.Thread(*args, **kwargs)
    t.daemon = True
    t.start()

    return t


def dumb_print(q):
    line = q.get()
    time.sleep(random.randint(1, 10))

    sys.stdout.write(line)

    q.task_done()


def main_loop():
    q = Queue.Queue(MAX_THREADS)

    with open('strings.txt', 'r') as f:
        data = f.readlines()

    for line in data:
        q.put(line)
        start_thread(target=dumb_print, args=(q,))

    q.join()


if __name__ == '__main__':
    sys.stdout.write('start\n')

    loop = start_thread(target=main_loop)

    while loop.is_alive():
        sys.stdout.write('still alive\n')
        time.sleep(random.randint(1, 5))

    sys.stdout.write('done\n')

Советую почитать документацию по Queue

p.s. как уже было упомянуто: если у вас вычисления на процессоре, а не IO, то лучше посмотреть в сторону модуля multiprocessing, благо что API почти не отличается.

Comments

[astrotrain]

Благодарю!

Answer 2

[bromzh]

Не надо писать многопоточные программы на питоне, потому что GIL.
Если нужна хорошая многопоточность - Java в руки.

Comments

[angru]

глупости(пруф), многопоточность имеет смысл, когда у нас много ввода/вывода. а еще есть модуль multiprocessing

[bromzh]

angru: вся статья наглядно иллюстрирует всю боль разработки чего-то параллельного на питоне. Процессы очень тяжелы, да и требуют хорошего процессора. Потоки нужно очень и очень хорошо уметь готовить, чтобы программы были хотя бы не медленнее их последовательных аналогов. Много I/O? Лучше тогда использовать asyncio. Или более подходящий язык.

[angru]

bromzh: ну да, так-то проще выучить новую экосистему, чем запомнить несколько нюансов использования потоков/процессов в питон.

[angru]

bromzh: asyncio медленный и только под третий питон, пока наверное все же лучше tornado или gevent(хотя я сейчас ковыряюсь в asyncio и собираюсь на нем по работе проектик запилить).

[astrotrain]

Хотелось бы по теме комментариев.

[angru]

astrotrain: для начала, нельзя изменять обьект во время итерации по нему, это я про:

for thr in threads:
    threads.remove(t)

[astrotrain]

angru: да, я нашел такую инфу, но как тогда удалить элемент из списка не создавая новый?

[eri]

bromzh: там есть такая штука как пул процессов - большие массивы разбираются очень быстро pool.map(....)