Питон-эквивалент выбора Голанга на каналах

Go имеет оператор выбора, который работает на каналах. Из документации:

Оператор select позволяет ждать goroutine в нескольких операциях связи.
Насколько хорошо началось?
Как сделать несколько аргументов для функции map, где один остается в python?
Определите, была ли удалена задача. GAE Python
Вход в facebook с помощью python
Можно ли прочитать файл с S3 в Google App Engine с помощью boto?

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

Есть ли эквивалент Python для следующего кода:

package main import "fmt" func main() { c1 := make(chan int) c2 := make(chan int) quit := make(chan int) go func() { for i := 0; i < 10; i++ { c1 <- i } quit <- 0 }() go func() { for i := 0; i < 2; i++ { c2 <- i } }() for { select { case <-c1: fmt.Println("Received value from c1") case <-c2: fmt.Println("Received value from c2") case <-quit: fmt.Println("quit") return } } }

Вывод этой программы:

 Received value from c1 Received value from c1 Received value from c2 Received value from c1 Received value from c2 Received value from c1 Received value from c1 Received value from c1 Received value from c1 Received value from c1 Received value from c1 Received value from c1 quit

6 Solutions collect form web for “Питон-эквивалент выбора Голанга на каналах”

Вот довольно прямой перевод, но «выбор, который, если несколько готовы», работает по-другому – это просто то, что было первым. Также это похоже на запуск вашего кода с помощью gomaxprocs(1) .

 import threading import Queue def main(): c1 = Queue.Queue(maxsize=0) c2 = Queue.Queue(maxsize=0) quit = Queue.Queue(maxsize=0) def func1(): for i in range(10): c1.put(i) quit.put(0) threading.Thread(target=func1).start() def func2(): for i in range(2): c2.put(i) threading.Thread(target=func2).start() combined = Queue.Queue(maxsize=0) def listen_and_forward(queue): while True: combined.put((queue, queue.get())) t = threading.Thread(target=listen_and_forward, args=(c1,)) t.daemon = True t.start() t = threading.Thread(target=listen_and_forward, args=(c2,)) t.daemon = True t.start() t = threading.Thread(target=listen_and_forward, args=(quit,)) t.daemon = True t.start() while True: which, message = combined.get() if which is c1: print 'Received value from c1' elif which is c2: print 'Received value from c2' elif which is quit: print 'Received value from quit' return main()

Основное изменение – это симуляция выбора с помощью потоков, которые объединяют сообщения. Если вы собираетесь использовать этот шаблон много, вы можете написать код выбора:

 import threading import Queue def select(*queues): combined = Queue.Queue(maxsize=0) def listen_and_forward(queue): while True: combined.put((queue, queue.get())) for queue in queues: t = threading.Thread(target=listen_and_forward, args=(queue,)) t.daemon = True t.start() while True: yield combined.get() def main(): c1 = Queue.Queue(maxsize=0) c2 = Queue.Queue(maxsize=0) quit = Queue.Queue(maxsize=0) def func1(): for i in range(10): c1.put(i) quit.put(0) threading.Thread(target=func1).start() def func2(): for i in range(2): c2.put(i) threading.Thread(target=func2).start() for which, msg in select(c1, c2, quit): if which is c1: print 'Received value from c1' elif which is c2: print 'Received value from c2' elif which is quit: print 'Received value from quit' return main()

Но…

Обратите внимание, что этот выбор не совсем подходит, хотя для вашей программы это не имеет значения – goroutine может отправить результат по каналу, который будет помещен в очередь в select и lost, если мы не всегда будем перебирать по выберите для завершения!

Также рассмотрите офсетную библиотеку Бенуа Чесно. Это порт параллельной модели Go для Python, используя волокна под обложками.

Он рассказал об этом на PyCon APAC 2013:

Слайды
видео

С Python 3.5 есть ключевые слова async и await которые позволяют иметь функции, которые могут быть приостановлены в процессе выполнения и, таким образом, могут работать на равномерном потоке вместо потоков. asyncio std lib предлагает один.

Чтобы более точно отобразить поведение каналов блокировки Go и select вы можете использовать эту небольшую библиотеку, а затем ваш примерный код будет очень похож на Python.

Вы можете использовать multiprocessing.Pipe вместо chan , threading.Thread вместо go и select.select вместо select .

Ниже приведен пример реализации вашего примера в Python с использованием этого подхода:

 import random from multiprocessing import Pipe from select import select from threading import Thread def main(): c1_r, c1_w = Pipe(duplex=False) c2_r, c2_w = Pipe(duplex=False) quit_r, quit_w = Pipe(duplex=False) def func1(): for i in range(10): c1_w.send(i) quit_w.send(0) Thread(target=func1).start() def func2(): for i in range(2): c2_w.send(i) Thread(target=func2).start() while True: ready, _, _ = select([c1_r, c2_r, quit_r], [], []) which = random.choice(ready) if which == c1_r: c1_r.recv() print 'Received value from c1' elif which == c2_r: c2_r.recv() print 'Received value from c2' elif which == quit_r: quit_r.recv() print 'Received value from quit' return if __name__ == '__main__': main()

Эта реализация основана на реализации @ Thomas, но в отличие от @ Thomas's она не порождает дополнительные потоки для выполнения выбора.

Протестировано на Linux с помощью Python 2.7.13. Windows может вести себя по-разному, так как выбор – это Unixy.

Вот еще одна попытка подражать синтаксису go:

 from threading import Thread from Queue import Queue def main(): c1 = Queue.Queue(maxsize=0) c2 = Queue.Queue(maxsize=0) quit = Queue.Queue(maxsize=0) Thread(target=lambda: [c1.put(i) for i in range(10)] or quit.put(0)).start() Thread(target=lambda: [c2.put(i) for i in range(2)]).start() for which, msg in select(c1, c2, quit): if which is c1: print 'Received value from c1' elif which is c2: print 'Received value from c2' elif which is quit: print 'Received value from quit' return def select(*queues): combined = Queue.Queue(maxsize=0) def listen_and_forward(queue): while True: combined.put((queue, queue.get())) for queue in queues: t = Thread(target=listen_and_forward, args=(queue,)) t.daemon = True t.start() while True: yield combined.get() main()

Да, все возможно с удовольствием . Можешь попробовать.

Повеселись 😉

Вот пример:

 c1 = goless.chan() c2 = goless.chan() def func1(): time.sleep(1) c1.send('one') goless.go(func1) def func2(): time.sleep(2) c2.send('two') goless.go(func2) for i in range(2): case, val = goless.select([goless.rcase(c1), goless.rcase(c2)]) print(val)