Существует ли многопоточная функция map()?

Ответ 1

Попробуйте функцию Pool.map из многопроцессорной обработки:

http://docs.python.org/library/multiprocessing.html#using-a-pool-of-workers

Это не многопоточно, но это действительно хорошо, поскольку многопоточность сильно искалечена на Python GIL.

Ответ 2

Вы можете использовать пакет многопроцессорности python (http://docs.python.org/library/multiprocessing.html). Пакет облачного питона, доступный из PiCloud (http://www.picloud.com), также предлагает функцию мультипроцессорной карты(), которая может разгрузить вашу к облаку.

Ответ 3

Теперь у Python есть модуль concurrent.futures, который является самым простым способом заставить карту работать с несколькими потоками или несколькими процессами.

https://docs.python.org/3/library/concurrent.futures.html

Ответ 4

Ниже моя функция map_parallel. Он работает так же, как map, за исключением того, что он может запускать каждый элемент параллельно в отдельном потоке (но см. Примечание ниже). Этот ответ основывается на другом SO-ответе.

import threading
import logging
def map_parallel(f, iter, max_parallel = 10):
    """Just like map(f, iter) but each is done in a separate thread."""
    # Put all of the items in the queue, keep track of order.
    from queue import Queue, Empty
    total_items = 0
    queue = Queue()
    for i, arg in enumerate(iter):
        queue.put((i, arg))
        total_items += 1
    # No point in creating more thread objects than necessary.
    if max_parallel > total_items:
        max_parallel = total_items

    # The worker thread.
    res = {}
    errors = {}
    class Worker(threading.Thread):
        def run(self):
            while not errors:
                try:
                    num, arg = queue.get(block = False)
                    try:
                        res[num] = f(arg)
                    except Exception as e:
                        errors[num] = sys.exc_info()
                except Empty:
                    break

    # Create the threads.
    threads = [Worker() for _ in range(max_parallel)]
    # Start the threads.
    [t.start() for t in threads]
    # Wait for the threads to finish.
    [t.join() for t in threads]

    if errors:
        if len(errors) > 1:
            logging.warning("map_parallel multiple errors: %d:\n%s"%(
                len(errors), errors))
        # Just raise the first one.
        item_i = min(errors.keys())
        type, value, tb = errors[item_i]
        # Print the original traceback
        logging.info("map_parallel exception on item %s/%s:\n%s"%(
            item_i, total_items, "\n".join(traceback.format_tb(tb))))
        raise value
    return [res[i] for i in range(len(res))]

ПРИМЕЧАНИЕ. Остерегайтесь исключения. Как и в случае с нормальным map, вышеприведенная функция вызывает исключение, если одна из ее подпоток вызывает исключение и останавливает итерацию. Однако из-за параллельного характера нет гарантии, что самый ранний элемент будет поднимать первое исключение.

Ответ 5

Возможно, попробуйте выполнить Внедрение Swallow Python 3? Это может быть крупный проект и не гарантированно стабильный, но если вы склонны, он может работать. Затем список или задание понятий кажется подходящей функциональной структурой для использования.

Ответ 6

Попробуйте concurrent.futures.ThreadPoolExecutor.map в стандартной библиотеке Python (новая версия 3.2).

Аналогично карте (func, * iterables) за исключением:

• итерируемые предметы собираются сразу, а не лениво;
• Функция func выполняется асинхронно, и несколько вызовов функции могут выполняться одновременно.

Простой пример (измененный из примера ThreadPoolExecutor):

import concurrent.futures
import urllib.request

URLS = [
  'http://www.foxnews.com/',
  'http://www.cnn.com/',
  'http://europe.wsj.com/',
  'http://www.bbc.co.uk/',
]

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    # Do something here
    # For example
    with urllib.request.urlopen(url, timeout=timeout) as conn:
      try:
        data = conn.read()
      except Exception as e:
        # You may need a better error handler.
        return b''
      else:
        return data

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
    # map
    l = list(executor.map(lambda url: load_url(url, 60), URLS))

print('Done.')

Ответ 7

Я бы подумал, что нет никакой причины иметь такую ​​функцию. Все потоки Python должны выполняться на одном процессоре. Предполагая, что ваша функция карты не имеет компонента ввода/вывода, вы не увидите ускорения при обработке (и, вероятно, увидите замедление из-за переключения контекста).

Другие плакаты упомянули многопроцессорность - это, вероятно, лучшая идея.

Ответ 8

Эта функциональность не встроена. Однако кто-то уже ее реализовал.