最新消息:20210917 已从crifan.com换到crifan.org

【已解决】Python中用多进程process测试单例Singleton

Python crifan 3575浏览 0评论

折腾:

【部分解决】Python中实现多线程或多进程中的单例singleton

期间,对于之前的ThreadSafeSingleton,虽然多线程thread测试是OK的,但是对于gunicorn的多worker是无效的。

gunicorn的多worker,目前的理解,应该属于多进程process。

所以也想想办法Mac本地能去调试多进程的情况,看看ThreadSafeSingleton是否有效。

python multiple process singleton

tendo/singleton.py at master · pycontribs/tendo

好像是支持多thread?

multiprocessing.Manager Python Example

Singleton Objects in Python – Michael Goerz

python singleton into multiprocessing – Stack Overflow

-》用shared memory去在多线程之间共享数据:

17.2. multiprocessing — Process-based parallelism — Python 3.7.0 documentation

python – Make Singleton class in Multiprocessing – Stack Overflow

python – State shared in multiprocessing when using singleton classes – Stack Overflow

16.6. multiprocessing — Process-based “threading” interface — Python 2.7.15 documentation

process – Sharing a complex object between Python processes? – Stack Overflow

17.2. multiprocessing — Process-based parallelism — Python 3.7.0 documentation

而对于如果真是多个process之间数据无法共享单例,那么暂时就放弃这条路,

暂时不去考虑用:shared memory,mmap等其他方式实现了,暂时还是考虑用:

【已解决】用gunicorn的gevent解决之前多worker多Process线程的单例的数据共享

的办法。

此处,先还是去参考:

https://github.com/pycontribs/tendo/blob/master/tendo/singleton.py

和:

17.2.1.4. Synchronization between processes

创建多个process去测试看看再说

期间对于Process类,有哪些函数和属性,参考:

17.2. multiprocessing — Process-based parallelism — Python 3.7.0 documentation

其中就有我要的:pid

"pid

Return the process ID. Before the process is spawned, this will be None."

以及:

https://docs.python.org/3/library/multiprocessing.html#multiprocessing.current_process

"multiprocessing.current_process()

Return the Process object corresponding to the current process.

An analogue of threading.current_thread()."

【总结】

最后用代码:

log.info("========== test multiple process singleton ==========")

import multiprocessing

def singleProcessDo(cur_num):

    log.info("———- singleProcessDo ———-")

    cur_process = multiprocessing.current_process()

    curProcessTTokenSingleton = MsTtsTokenSingleton()

    log.info("curProcessTTokenSingleton=%s", curProcessTTokenSingleton)

    log.info("[%d] name=%s, pid=%s, process=%s", cur_num, cur_process.name, cur_process.pid, cur_process)

max_process_num = 5

for idx in range(max_process_num):

    cur_num = idx + 1

    each_process_name = "P%s" % cur_num

    cur_process = multiprocessing.Process(target=singleProcessDo, name=each_process_name, args=(cur_num, ))

    log.info("[%d] name=%s, process=%s", cur_num, each_process_name, cur_process)

    cur_process.start()

测试结果竟然是:

多Process进程中,对应的单例,也是工作的:

【后记】

但是gunicorn中的多个worker中单例,就是不工作的:还是会初始化多个实例的。

最终发现是另外的原因:

【未解决】Flask中gunicorn部署和supervisor管理celery的worker导致多线程导致单例失效

转载请注明:在路上 » 【已解决】Python中用多进程process测试单例Singleton

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
92 queries in 0.201 seconds, using 23.40MB memory