python - 在 __init__ 类中初始化多进程是否正确？

问题描述

from multiprocessing.dummy import Pool as ThreadPool
class TSNew:
    def __init__(self):
        self.redis_client = redis.StrictRedis(host="172.17.31.147", port=4401, db=0)
        self.global_switch = 0
        self.pool = ThreadPool(40) # init pool
        self.dnn_model = None
        self.nnf = None
        self.md5sum_nnf = "initialize"
        self.thread = threading.Thread(target=self.load_model_item)
        self.ts_picked_ids = None
        self.thread.start()

        self.memory = deque(maxlen=3000)
        self.process = threading.Thread(target=self.process_user_dict)
        self.process.start()

    def load_model_item(self):
        '''
        code
        '''
    def predict_memcache(self,user_dict):
        '''
        code
        '''
    def process_user_dict(self):
        while True:
            '''
            code to generate user_dicts which is a list 
            '''
            results = self.pool.map(self.predict_memcache, user_dicts)
            '''
            code
            '''
TSNew_ = TSNew()

def get_user_result():
    logging.info("----------------come in ------------------")
    if request.method == 'POST':
        user_dict_json = request.get_data()# userid
        if user_dict_json == '' or user_dict_json is None:
            logging.info("----------------user_dict_json is ''------------------")
            return ''
        try:
            user_dict = json.loads(user_dict_json)
        except:
            logging.info("json load error, pass")
            return ''
        TSNew_.memory.append(user_dict)
        logging.info('add to deque TSNew_.memory size: %d  PID: %d', len(TSNew_.memory), os.getpid())
        logging.info("add to deque userid: %s, nation: %s \n",user_dict['user_id'],  user_dict['user_country'])
        return 'SUCCESS\n'


@app.route('/', methods=['POST'])
def get_ts_gbdt_id():
    return get_user_result()

from werkzeug.contrib.fixers import ProxyFix
app.wsgi_app = ProxyFix(app.wsgi_app)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=4444)

我在类中创建了一个多线程池，并__init__使用. 我有两个疑问：（a）我应该初始化池还是在之前初始化它self.poolpredict_memcache__init__

results = self.pool.map(self.predict_memcache, user_dicts)

(b) 由于pool是多线程操作，并且是在线程中执行的process_user_dict，所以有没有隐藏的错误？谢谢。

标签： pythonmultiprocessingthread-safety

解决方案

问题（一）：

这取决于。如果您需要多次运行process_user_dict，那么在构造函数中启动池并使其保持运行是有意义的。创建线程池总是会带来一些开销，通过在调用之间保持池处于活动状态process_user_dict可以避免额外的开销。

如果您只想处理一组输入，您也可以在process_user_dict. 但之前可能不会，因为这会为您周围循环results = self.pool.map(self.predict_memcache, user_dicts)的每次迭代创建一个池。while

在您的具体情况下，它没有任何区别。您TSNew_在模块级别创建对象，以便在您的应用程序运行时它保持活动状态（以及线程池）；来自同一TSNew实例的同一线程池用于处理app.run(). 由于您似乎将该构造self.process = threading.Thread(target=self.process_user_dict)用作某种侦听器 on self.memory，因此在构造函数中创建池在功能上等同于在内部process_user_dict（但在循环外部）创建池。

问题（b）：

从技术上讲，在线程内部创建线程时，默认情况下没有隐藏错误。最后，任何附加线程的最终父级始终是MainThread，它为 Python 解释器的每个实例隐式创建。基本上，每次在 Python 程序中创建线程时，都会在线程中创建线程。

实际上，您的代码甚至没有在线程内创建线程。你self.pool是在MainThread. 当池通过self.pool = ThreadPool(40)它实例化时，它会创建所需数量 (40) 的工作线程，加上一个工作处理程序线程、一个任务处理程序线程和一个结果处理程序线程。所有这些都是MainThread. 关于线程内的池，您所做的self.process只是调用它的map方法来为其分配任务。

但是，我并没有真正看到您在这里所做的事情的意义self.process。猜测一下，我会说您想启动循环 inprocess_user_dict以充当上的侦听器self.memory，以便池user_dict在它们开始出现在 in 时立即deque开始处理self.memory。从我看到你在做的事情来看get_user_result，你似乎user_dict每个请求都会得到一个。我知道您可能有并发用户会话传入这些 dicts，但是您真的看到process_user_dict在无限循环中运行而不是简单地调用TSNew_.process_user_dict()after的好处TSNew_.memory.append(user_dict)吗？您甚至可以完全省略self.memory并将 dict 直接传递给process_user_dict，除非我遗漏了您没有向我们展示的内容。

python - 在 init 类中初始化多进程是否正确？

问题描述

解决方案

推荐阅读