如何在 Flask 中并行化任务？

Question

我正在向我的 Flask 服务器发送 XHR 请求，以便在网络上执行多次 ping 操作

资源

def get(self, site_id):
    …
    for printer in printers:
        hostname = printer['hostname']
        response[site_id][hostname] = network_utils.ping(hostname)

    return response

平

在shell.execute下面我用subprocess.check_output来运行一个原生的ping：

def ping(hostname):
    command = ['ping', '-c', '1', '-W', '1', '-q', hostname]

    response = shell.execute(command)
    return output_parser.ping(response['results'])

输出

{
    "test-site": {
        "avg": 0.093, "max": 0.093, "mdev": 0.0, "min": 0.093,
        "1.1.1.1": { "avg": null, "max": null, "mdev": null, "min": null},
        "1.2.3.4": { "avg": null, "max": null, "mdev": null, "min": null},
        "127.0.0.1": { "avg": 0.061, "max": 0.061, "mdev": 0.0, "min": 0.061}
    }
}

问题

ping 是按顺序运行的，使得请求超级慢（几十秒，我怎样才能加快速度？

Answer 1

听起来最好的选择是线程，因为您的问题是I/O 绑定。我正在使用Semaphore 限制为 5 个线程。

我将响应 dict 发送到 ping dict 是线程安全的，但如果您考虑更复杂的事情，您应该阅读 this。

def get(self, site_id):
    …
    semaphore = threading.Semaphore(5)
    threads = []

    for printer in printers:
        hostname = printer['hostname']
        threads.append(threading.Thread(target=network_utils.ping,
                          args=(semaphore, response, site_id, hostname)))

    # Start and wait to all threads to finish
    map(lambda t: t.start(), threads)
    map(lambda t: t.join(), threads)

    return response

def ping(semaphore, response, site_id, hostname):
    semaphore.acquire()

    command = ['ping', '-c', '1', '-W', '1', '-q', hostname]
    response = shell.execute(command)
    ping_data = output_parser.ping(response['results'])

    response[site_id][hostname] = ping_data

    semaphore.release()

Answer 2

例如，通过gevent 使子进程异步。

from gevent import subprocess
import gevent

def ping(hostname):
    command = ['ping', '-c', '1', '-W', '1', '-q', hostname]
    return subprocess.Popen(command, stdout=subprocess.PIPE)

def get(self, site_id):
    …
    # Start all the pings in parallel, asynchronously
    # Use dict to reference host: ping subprocess
    # as results will come in at different times
    pings = {printer['hostname']: ping(printer['hostname']) 
             for printer in printers}
    # Wait for all of them to complete
    gevent.wait(pings.values())
    for hostname in pings:
        response[site_id][hostname] = output_parser.ping(pings[hostname].stdout.read())
    return response

Answer 3

Upvote Or Duan answer as mine 是基于他的回答：

资源

class Ping(Resource):
    def get(self, site_id):
        site_hostname = mast_utils.list_sites(site_id)['results'][0]['hostname']
        printers = mast_utils.list_printers(site_id)['results']['channels']

        response = network_utils.parellelize(network_utils.ping, site_hostname, printers)
        return response

api.add_resource(Ping, '/ping/<string:site_id>/')

network_utils.py

def ping(hostname):
    command = ['ping', '-q', hostname,
               '-w', '1',
               '-W', '1',
               '-i', '0.2'
               ]

    response = shell.execute(command)

    return output_parser.ping(response['results'])


def collect(task, response, **kwargs):
    hostname = kwargs['hostname']

    response[hostname] = task(**kwargs)


def parellelize(task, site_id, printers, **kwargs):
    response = {}
    kw = kwargs.copy()
    kw.update({'hostname': site_id})
    collect(task, response, **kw)

    printers_response = {}
    threads = []
    for printer in printers:
        hostname = printer['hostname']
        kw = kwargs.copy()
        kw.update({'hostname': hostname})

        threads.append(
            threading.Thread(
                target=collect,
                args=(task, printers_response),
                kwargs=kw
            )
        )

    for thread in threads:
        thread.start()
        thread.join()

    response[site_id].update(printers_response)

    return response

test_network_utils.py

class NetwrokUtilsTestCase(unittest.TestCase):
    def test_ping_is_null_when_host_unreachable(self):
        hostname = 'unreachable'

        response = network_utils.ping(hostname)

        self.assertDictEqual(response, {
            'avg': None,
            'max': None,
            'mdev': None,
            'min': None
        })

    def test_ping_reply_time_when_reachable(self):
        hostname = '127.0.0.1'

        response = network_utils.ping(hostname)

        self.assertGreater(response['avg'], 0)

    def test_ping_with_only_a_site(self):
        site_hostname = 'localhost'
        printers = []
        response = {}

        response = network_utils.parellelize(network_utils.ping, site_hostname, printers)

        self.assertGreater(response[site_hostname]['avg'], 0)

    def test_ping_with_printers(self):
        site_hostname = 'localhost'
        printers = [
            {'hostname': '127.0.0.1', 'port': 22},
            {'hostname': '0.0.0.0', 'port': 22},
        ]

        response = network_utils.parellelize(network_utils.ping, site_hostname, printers)

        self.assertGreater(response[site_hostname]['avg'], 0)
        self.assertGreater(response[site_hostname]['127.0.0.1']['avg'], 0)