Как передавать / обмениваться данными из асинхронных методов / потоков? - PullRequest
Новая
       19

Как передавать / обмениваться данными из асинхронных методов / потоков?

0 голосов
/ 09 января 2019

Может ли кто-нибудь объяснить концепцию передачи данных между асинхронными вызовами в python?

У меня есть такой сценарий:
У меня есть основная процедура (mainthread), а затем я запускаю другой асинхронный вызов, который добавляет два числа () и спит в течение некоторого времени. Ожидается, что основной поток будет ждать, пока не завершится calc и sleep. С точки зрения псевдокода это может выглядеть так: 

def example():
    def my_calc(x,y):
        z = x+ y
        time.sleep(2)
        return z  #this should get pushed to queue, but for simplicity we use 'return z'

    z = asyncSomething.callInThread(my_calc, 2, 20)  #assume we get z from queue
    return z+10

def runner():  #our main function
    print 'start'
    z = example()
    print 'result is {0}'.format(z)

Как я заставил последние print ждать z? Я пытался использовать threading.Event () и играл с set, wait и clear, но они блокируют все.

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

Я думаю, что мне не хватает реальной концепции асинхронной работы в python.

Ответы [ 2 ]

0 голосов
/ 09 января 2019

Если ваш основной поток продвинулся настолько далеко, насколько это возможно без результата рабочего потока, и вы хотите заблокировать его, вы можете использовать threading.Event для ожидания определенного момента в выполнении рабочего потока или thread.join (где thread - экземпляр threading.Thread), чтобы дождаться завершения рабочего потока: 

class Example(object):
    def my_calc(self, x, y):
        time.sleep(2)
        self._z = x + y # or push the result to a queue

    def example(self):
        thread = threading.Thread(target=self.my_calc, args=(2, 20))
        thread.start()
        # any other useful work could go here while the worker runs
        thread.join()
        return self._z + 10 # or grab the result from a queue

def runner():
    print "start"
    z = Example().example()
    print "result is {0}".format(z)

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

def my_calc(x, y):
    time.sleep(2)
    return x + y

def my_calc_thread(x, y, callback):
    z = my_calc(x, y)
    # depending on your GUI framework, you may need to do something
    # like call_in_main_thread(callback, z) if callback touches GUI
    # elements
    callback(z)

def example():
    def finish(z):
        print "result is {0}".format(z)
    t = threading.Thread(target=my_calc_thread, args=(2, 20, finish))
    t.start()

def runner():
    print "start"
    example()
0 голосов
/ 09 января 2019

Если вы используете многопоточность, вы должны использовать futures для инкапсуляции ваших результатов. В частности, future.result(timeout=None) будет использоваться в вашей ситуации:

Возвращает значение, возвращаемое вызовом. Если вызов еще не завершен, этот метод будет ждать до тайм-аут секунд. Если вызов не был завершен в течение тайм-аута, то будет вызвано concurrent.futures.TimeoutError. Тайм-аут может быть int или float. Если время ожидания не указано или отсутствует, время ожидания не ограничено.

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

...