python implements blocking mode method resolution for class AutoResetEvent

  • 2020-05-09 19:04:35
  • OfStack

Those of you who have done C# multithreading understand both the AutoResetEvent and ManualResetEvent classes, where the WaitOne() method and the Set() and Reset() methods are used more in thread synchronization.
AutoResetEvent: when a thread executes to the WaitOne() method, the thread will be in blocking mode. When the Set() method is called, the blocked thread will continue to execute, and its state will be automatically set to blocking mode.
ManualResetEvent: when a thread executes to the WaitOne() method, it will be in blocking mode. When the Set() method is called, the blocked thread will continue to execute down. Its state will not be automatically set to blocking mode, but must be called to the Reset() method to set its state to blocking mode.

There is also a similar thread blocking mode class threading.Event () under the threading module in python, which is similar to ManualResetEvent in C# and does not automatically change to blocking mode after calling the set() method. Sometimes we need this automatic blocking mode in the development project. I have encountered this kind of demand in the project, so I tried to write a similar class by myself. Please post the code and share it.
The code is as follows:
 
# encoding: UTF-8 
import threading 
class AutoEvent: 
def __init__(self): 
self.event = None 
self.is_wait = False #  Whether it's blocked or not  
def Wait(self,timeout=None): 
if not self.is_wait: 
self.is_wait = True 
self.event = threading.Event() #  instantiation threading.Event() object  
self.event.wait(timeout=timeout) #  call threading.Event().wait() Method to block the thread  
del self.event #  Release object  
def Set(self): 
if not self.is_wait:raise 'must be invoke Wait() method before Set()' 
self.is_wait = False 
self.event.set() #  call threading.Event().set() Method to let the thread continue  

Here's Demo for the test:
 
import wx 
autoEvent = AutoEvent() #  instantiation MamualEvent object  
class TestDialog(wx.Dialog): 
def __init__(self, parent): 
wx.Dialog.__init__(self, parent, -1) 
b = wx.Button(self, -1, "print", (50, 140)) 
self.Bind(wx.EVT_BUTTON, self.OnPrint, b) 
self.thread = threading.Thread(target=self.work) 
self.thread.start() 
def work(self): 
while True: 
print 3 
autoEvent.Wait() #  blocking  
def OnPrint(self, evt): 
autoEvent.Set() #  Continue to  
if __name__=="__main__": 
app = wx.App() 
f = TestDialog(parent=None) 
f.Show() 
app.MainLoop() 

When the program is executed, a '3' will be printed first, and a '3' will be printed without clicking the print button.
The test was successful.
Note: if you change the autoEvent.Wait () and autoEvent.Set () in the Demo code to threading.Event (), wait() and set() methods in the threading.Event () class, the result will be as follows: when the program is executed, a '3' will be printed.

Related articles: