在Python中正确使用互斥锁

Question

我开始使用python中的多线程(或者至少我的脚本可能会创建多个线程).这个算法是否是Mutex的正确用法？我还没有测试过这段代码,它甚至可能都无法运行.我只是希望processData在一个线程中运行(一次一个)并且主while循环继续运行,即使队列中有一个线程.

from threading import Thread
from win32event import CreateMutex
mutex = CreateMutex(None, False, "My Crazy Mutex")
while(1)
    t = Thread(target=self.processData, args=(some_data,))
    t.start()
    mutex.lock()

def processData(self, data)
    while(1)
        if mutex.test() == False:
            do some stuff
            break

编辑:重新阅读我的代码我可以看到它是非常错误的.但是,嘿,这就是我在这里寻求帮助的原因.

Answer 1

我不知道你为什么使用Window的Mutex而不是Python的.使用Python方法,这非常简单:

from threading import Thread, Lock

mutex = Lock()

def processData(data):
    mutex.acquire()
    try:
        print('Do some stuff')
    finally:
        mutex.release()

while True:
    t = Thread(target = processData, args = (some_data,))
    t.start()

但是请注意,由于CPython的体系结构(即全局解释器锁),你实际上一次只能运行一个线程 - 如果它们中的一些是I/O绑定的话,这很好,尽管你需要尽可能释放锁,以便I/O绑定线程不会阻止其他线程运行.

对于Python 2.6及更高版本,另一种方法是使用Python的multiprocessing包.它反映了threading包,但将创建可以同时运行的全新流程.更新您的示例是微不足道的:

from multiprocessing import Process, Lock

mutex = Lock()

def processData(data):
    with mutex:
        print('Do some stuff')

if __name__ == '__main__':
    while True:
        p = Process(target = processData, args = (some_data,))
        p.start()

Answer 2

这是我提出的解决方案:

import time
from threading import Thread
from threading import Lock

def myfunc(i, mutex):
    mutex.acquire(1)
    time.sleep(1)
    print "Thread: %d" %i
    mutex.release()


mutex = Lock()
for i in range(0,10):
    t = Thread(target=myfunc, args=(i,mutex))
    t.start()
    print "main loop %d" %i

输出:

main loop 0
main loop 1
main loop 2
main loop 3
main loop 4
main loop 5
main loop 6
main loop 7
main loop 8
main loop 9
Thread: 0
Thread: 1
Thread: 2
Thread: 3
Thread: 4
Thread: 5
Thread: 6
Thread: 7
Thread: 8
Thread: 9

Answer 3

你必须在某个时候解锁你的Mutex ......

Answer 4

我想改善的答案从克里斯-B 多一点点.

请参阅下面的代码:

from threading import Thread, Lock
import threading
mutex = Lock()


def processData(data, thread_safe):
    if thread_safe:
        mutex.acquire()
    try:
        thread_id = threading.get_ident()
        print('\nProcessing data:', data, "ThreadId:", thread_id)
    finally:
        if thread_safe:
            mutex.release()


counter = 0
max_run = 100
thread_safe = False
while True:
    some_data = counter        
    t = Thread(target=processData, args=(some_data, thread_safe))
    t.start()
    counter = counter + 1
    if counter >= max_run:
        break

在你的第一次运行中,如果你thread_safe = False在while循环中设置,则不会使用互斥锁,并且线程将在print方法中相互跳过,如下所示;

但是,如果你设置thread_safe = True并运行它,你会看到所有输出完全正常;

希望这可以帮助.