在异步/多线程中处理可变共享对象。替代'.copy（）'？

Question

我正在编写一个程序，它创建了一堆工人来使用aiohttp异步调用API。但是这个问题是关于共享对象的。我假设如果我是多线程的话，我会遇到相同或类似的问题。

我有一组所有工作人员共享的默认URL参数，但是这些参数的两个值从一个工人改为工人：

DEFAULT_PARAMS = {
    'q' : None,                         #<==CHANGES per worker
    'offset' : '0',                     #<==CHANGES per worker
    'mkt' : 'en-US',                    #<==STATIC for all workers
    'moreParams' : '<most of the data>' #<==STATIC for all workers
}  

这是我如何初始化我的Worker()类：

class Worker(object):
    def __init__(self, q):
        # this copy iexpensive when > 100 workers.
        self.initial_params = DEFAULT_PARAMS.copy()
        # but witout copying entire default params dict, the next line
        # would add alter the 'q' value for all instances of Worker.
        self.initial_params.update({'q' : q})

我正在寻找一种替代方法来为我创建的每个新工人调用DEFAULT_PARAMS.copy()。

弄清楚如何提出这个问题一直是一个挑战。我怀疑我的答案可能在课堂上的某个地方通过实例属性。

这是我的程序非常简单的例子：

import aiohttp
import asyncio

DEFUALT_PARAMS = {
    'q' : None, #<==CHANGES per worker
    'offset' : '0', #<==CHANGES per worker
    'mkt' : 'en-US', #<==STATIC for all workers
    'moreParams' : '<most of the data>' #<==STATIC for all workers
}

class Worker(object):
    def __init__(self, q):
        self.initial_params = DEFUALT_PARAMS.copy() # <==expensive
        self.initial_params.update({'q' : q}) #<==without copying, overwrites ref for all classes.

    async def call_api(self):
        async with aiohttp.ClientSession() as sesh:
            async with sesh.get(
                'https://somesearchengine.com/search?',
                params=self.initial_params
            ) as resp:
                assert resp.status == 200
                print(await resp.json())


async def main(workers, *, loop=None):
    tasks = (asyncio.ensure_future(i.call_api(), loop=loop) for i in workers)
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queries = ['foo', 'bar', 'baz']
    workers = (Worker(i) for i in queries)
    loop.run_until_complete(main(workers, loop=loop))

Answer 1

复制字典，即使是100名工人，也不是那么昂贵。您可以在7微秒内创建1000键字典的副本并进行更新：

>>> from timeit import Timer
>>> from secrets import token_urlsafe
>>> test_dict = {token_urlsafe(): token_urlsafe() for _ in range(1000)}
>>> len(test_dict)
1000
>>> count, total = Timer("p = d.copy(); p.update({'q' : q})", "from __main__ import test_dict as d; q = 42").autorange()
>>> print(total/count*1000000)  # microseconds are 10**-6 seconds
7.146239580000611

所以我要说这里没有问题。

但是，您实际上是在分层字典内容;将每个工人调整为一个或两个键。您可以使用collections.ChainMap() object来处理分层，而不是创建副本。 ChainMap()对象需要多个字典，并在这些字典中查找键，直到找到值。没有创建副本，并且当您改变地图时，最顶层的字典用于设置值：

from collections import ChainMap

# ...
self.initial_params = ChainMap({'q': q}, DEFAULT_PARAMS)

创建ChainMap()对象仍然更便宜：

>>> count, total = Timer("p = ChainMap({'q': q}, d)", "from __main__ import test_dict as d; q = 42; from collections import ChainMap").autorange()
>>> print(total/count*1000000)
0.5310121239999717

所以只有半微秒。当然，这需要以较慢的迭代和每个密钥访问为代价。这将取决于aiohttp如何处理这些，我建议您使用timeit模块执行自己的微基准测试，以测量代码执行的实际操作的性能。

但是请注意，在尝试使用任何并发模型处理共享状态时，总是要付出代价，并且即使没有并发性，在实例之间共享字典也总是存在问题。

Answer 2

如果q由工人拥有，为什么不在工作者本身中将其作为实例变量。

class Worker(object):
    def __init__(self, q):
      self.q = q

无论你想要什么q做self.q