快捷導(dǎo)航

分析Python編程時(shí)利用wxPython來(lái)支持多線程的方法

更新時(shí)間：2015年04月07日 15:50:18 投稿：goldensun

這篇文章主要介紹了Python編程時(shí)利用wxPython來(lái)支持多線程的方法,本文主要以開(kāi)發(fā)GUI程序時(shí)做線程通訊作為一個(gè)示例來(lái)講解,需要的朋友可以參考下

如果你經(jīng)常使用python開(kāi)發(fā)GUI程序的話，那么就知道，有時(shí)你需要很長(zhǎng)時(shí)間來(lái)執(zhí)行一個(gè)任務(wù)。當(dāng)然，如果你使用命令行程序來(lái)做的話，你回非常驚訝。大部分情況下，這會(huì)堵塞GUI的事件循環(huán)，用戶(hù)會(huì)看到程序卡死。如何才能避免這種情況呢？當(dāng)然是利用線程或進(jìn)程了！本文，我們將探索如何使用wxPython和theading模塊來(lái)實(shí)現(xiàn)。

wxpython線程安全方法

wxPython中，有三個(gè)“線程安全”的函數(shù)。如果你在更新UI界面時(shí)，三個(gè)函數(shù)都不使用，那么你可能會(huì)遇到奇怪的問(wèn)題。有時(shí)GUI也忙運(yùn)行挺正常，有時(shí)卻會(huì)無(wú)緣無(wú)故的崩潰。因此就需要這三個(gè)線程安全的函數(shù)：wx.PostEvent, wx.CallAfter和wx.CallLater。據(jù)Robin Dunn（wxPython作者）描述，wx.CallAfter使用了wx.PostEvent來(lái)給應(yīng)用程序?qū)ο蟀l(fā)生事件。應(yīng)用程序會(huì)有個(gè)事件處理程序綁定到事件上，并在收到事件后，執(zhí)行處理程序來(lái)做出反應(yīng)。我認(rèn)為wx.CallLater是在特定時(shí)間后調(diào)用了wx.CallAfter函數(shù)，已實(shí)現(xiàn)規(guī)定時(shí)間后發(fā)送事件。

Robin Dunn還指出Python全局解釋鎖 (GIL)也會(huì)避免多線程同時(shí)執(zhí)行python字節(jié)碼，這會(huì)限制程序使用CPU內(nèi)核的數(shù)量。另外，他還說(shuō)，“wxPython發(fā)布GIL是為了在調(diào)用wx API時(shí)，其他線程也可以運(yùn)行”。換句話說(shuō)，在多核機(jī)器上使用多線程，可能效果會(huì)不同。

總之，大概的意思是桑wx函數(shù)中，wx.CallLater是最抽象的線程安全函數(shù), wx.CallAfter次之，wx.PostEvent是最低級(jí)的。下面的實(shí)例，演示了如何使用wx.CallAfter和wx.PostEvent函數(shù)來(lái)更新wxPython程序。

wxPython, Theading, wx.CallAfter and PubSub

wxPython郵件列表中，有些專(zhuān)家會(huì)告訴其他人使用wx.CallAfter，并利用PubSub實(shí)現(xiàn)wxPython應(yīng)用程序與其他線程進(jìn)行通訊，我也贊成。如下代碼是具體實(shí)現(xiàn)：

import time 
import wx 
   
from threading import Thread 
from wx.lib.pubsub import Publisher 
   
######################################################################## 
class TestThread(Thread): 
  """Test Worker Thread Class."""
   
  #---------------------------------------------------------------------- 
  def __init__(self): 
    """Init Worker Thread Class."""
    Thread.__init__(self) 
    self.start()  # start the thread 
   
  #---------------------------------------------------------------------- 
  def run(self): 
    """Run Worker Thread."""
    # This is the code executing in the new thread. 
    for i in range(6): 
      time.sleep(10) 
      wx.CallAfter(self.postTime, i) 
    time.sleep(5) 
    wx.CallAfter(Publisher().sendMessage, "update", "Thread finished!") 
   
  #---------------------------------------------------------------------- 
  def postTime(self, amt): 
    """
    Send time to GUI
    """
    amtOfTime = (amt + 1) * 10
    Publisher().sendMessage("update", amtOfTime) 
   
######################################################################## 
class MyForm(wx.Frame): 
   
  #---------------------------------------------------------------------- 
  def __init__(self): 
    wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") 
   
    # Add a panel so it looks the correct on all platforms 
    panel = wx.Panel(self, wx.ID_ANY) 
    self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here") 
    self.btn = btn = wx.Button(panel, label="Start Thread") 
   
    btn.Bind(wx.EVT_BUTTON, self.onButton) 
   
    sizer = wx.BoxSizer(wx.VERTICAL) 
    sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5) 
    sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) 
    panel.SetSizer(sizer) 
   
    # create a pubsub receiver 
    Publisher().subscribe(self.updateDisplay, "update") 
   
  #---------------------------------------------------------------------- 
  def onButton(self, event): 
    """
    Runs the thread
    """
    TestThread() 
    self.displayLbl.SetLabel("Thread started!") 
    btn = event.GetEventObject() 
    btn.Disable() 
   
  #---------------------------------------------------------------------- 
  def updateDisplay(self, msg): 
    """
    Receives data from thread and updates the display
    """
    t = msg.data 
    if isinstance(t, int): 
      self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) 
    else: 
      self.displayLbl.SetLabel("%s" % t) 
      self.btn.Enable() 
   
#---------------------------------------------------------------------- 
# Run the program 
if __name__ == "__main__": 
  app = wx.PySimpleApp() 
  frame = MyForm().Show() 
  app.MainLoop()

我們會(huì)用time模塊來(lái)模擬耗時(shí)過(guò)程，請(qǐng)隨意將自己的代碼來(lái)代替，而在實(shí)際項(xiàng)目中，我用來(lái)打開(kāi)Adobe Reader，并將其發(fā)送給打印機(jī)。這并沒(méi)什么特別的，但我不用線程的話，應(yīng)用程序中的打印按鈕就會(huì)在文檔發(fā)送過(guò)程中卡住，UI界面也會(huì)被掛起，直到文檔發(fā)送完畢。即使一秒，兩秒對(duì)用戶(hù)來(lái)說(shuō)都有卡的感覺(jué)。

總之，讓我們來(lái)看看是如何工作的。在我們編寫(xiě)的Thread類(lèi)中，我們重寫(xiě)了run方法。該線程在被實(shí)例化時(shí)即被啟動(dòng)，因?yàn)槲覀冊(cè)赺_init__方法中有“self.start”代碼。run方法中，我們循環(huán)6次，每次sheep10秒，然后使用wx.CallAfter和PubSub更新UI界面。循環(huán)結(jié)束后，我們發(fā)送結(jié)束消息給應(yīng)用程序，通知用戶(hù)。

你會(huì)注意到，在我們的代碼中，我們是在按鈕的事件處理程序中啟動(dòng)的線程。我們還禁用按鈕，這樣就不能開(kāi)啟多余的線程來(lái)。如果我們讓一堆線程跑的話，UI界面就會(huì)隨機(jī)的顯示“已完成”，而實(shí)際卻沒(méi)有完成，這就會(huì)產(chǎn)生混亂。對(duì)用戶(hù)來(lái)說(shuō)是一個(gè)考驗(yàn)，你可以顯示線程PID，來(lái)區(qū)分線程，你可能要在可以滾動(dòng)的文本控件中輸出信息，這樣你就能看到各線程的動(dòng)向。

最后可能就是PubSub接收器和事件的處理程序了：

def updateDisplay(self, msg): 
  """
  Receives data from thread and updates the display
  """
  t = msg.data 
  if isinstance(t, int): 
    self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) 
  else: 
    self.displayLbl.SetLabel("%s" % t) 
    self.btn.Enable()

看我們?nèi)绾螐木€程中提取消息，并用來(lái)更新界面？我們還使用接受到數(shù)據(jù)的類(lèi)型來(lái)告訴我們什么顯示給了用戶(hù)。很酷吧？現(xiàn)在，我們玩點(diǎn)相對(duì)低級(jí)一點(diǎn)點(diǎn)，看wx.PostEvent是如何辦的。

wx.PostEvent與線程

下面的代碼是基于wxPython wiki編寫(xiě)的，這看起來(lái)比wx.CallAfter稍微復(fù)雜一下，但我相信我們能理解。

import time 
import wx 
   
from threading import Thread 
   
# Define notification event for thread completion 
EVT_RESULT_ID = wx.NewId() 
   
def EVT_RESULT(win, func): 
  """Define Result Event."""
  win.Connect(-1, -1, EVT_RESULT_ID, func) 
   
class ResultEvent(wx.PyEvent): 
  """Simple event to carry arbitrary result data."""
  def __init__(self, data): 
    """Init Result Event."""
    wx.PyEvent.__init__(self) 
    self.SetEventType(EVT_RESULT_ID) 
    self.data = data 
   
######################################################################## 
class TestThread(Thread): 
  """Test Worker Thread Class."""
   
  #---------------------------------------------------------------------- 
  def __init__(self, wxObject): 
    """Init Worker Thread Class."""
    Thread.__init__(self) 
    self.wxObject = wxObject 
    self.start()  # start the thread 
   
  #---------------------------------------------------------------------- 
  def run(self): 
    """Run Worker Thread."""
    # This is the code executing in the new thread. 
    for i in range(6): 
      time.sleep(10) 
      amtOfTime = (i + 1) * 10
      wx.PostEvent(self.wxObject, ResultEvent(amtOfTime)) 
    time.sleep(5) 
    wx.PostEvent(self.wxObject, ResultEvent("Thread finished!")) 
   
######################################################################## 
class MyForm(wx.Frame): 
   
  #---------------------------------------------------------------------- 
  def __init__(self): 
    wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") 
   
    # Add a panel so it looks the correct on all platforms 
    panel = wx.Panel(self, wx.ID_ANY) 
    self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here") 
    self.btn = btn = wx.Button(panel, label="Start Thread") 
   
    btn.Bind(wx.EVT_BUTTON, self.onButton) 
   
    sizer = wx.BoxSizer(wx.VERTICAL) 
    sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5) 
    sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) 
    panel.SetSizer(sizer) 
   
    # Set up event handler for any worker thread results 
    EVT_RESULT(self, self.updateDisplay) 
   
  #---------------------------------------------------------------------- 
  def onButton(self, event): 
    """
    Runs the thread
    """
    TestThread(self) 
    self.displayLbl.SetLabel("Thread started!") 
    btn = event.GetEventObject() 
    btn.Disable() 
   
  #---------------------------------------------------------------------- 
  def updateDisplay(self, msg): 
    """
    Receives data from thread and updates the display
    """
    t = msg.data 
    if isinstance(t, int): 
      self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) 
    else: 
      self.displayLbl.SetLabel("%s" % t) 
      self.btn.Enable() 
   
#---------------------------------------------------------------------- 
# Run the program 
if __name__ == "__main__": 
  app = wx.PySimpleApp() 
  frame = MyForm().Show() 
  app.MainLoop()

讓我們先稍微放一放，對(duì)我來(lái)說(shuō)，最困擾的事情是第一塊：

# Define notification event for thread completion 
EVT_RESULT_ID = wx.NewId() 
   
def EVT_RESULT(win, func): 
  """Define Result Event."""
  win.Connect(-1, -1, EVT_RESULT_ID, func) 
   
class ResultEvent(wx.PyEvent): 
  """Simple event to carry arbitrary result data."""
  def __init__(self, data): 
    """Init Result Event."""
    wx.PyEvent.__init__(self) 
    self.SetEventType(EVT_RESULT_ID) 
    self.data = data

EVT_RESULT_ID只是一個(gè)標(biāo)識(shí)，它將線程與wx.PyEvent和“EVT_RESULT”函數(shù)關(guān)聯(lián)起來(lái)，在wxPython代碼中，我們將事件處理函數(shù)與EVT_RESULT進(jìn)行捆綁，這就可以在線程中使用wx.PostEvent來(lái)將事件發(fā)送給自定義的ResultEvent了。

結(jié)束語(yǔ)

希望你已經(jīng)明白在wxPython中基本的多線程技巧。還有其他多種多線程方法這里就不在涉及，如wx.Yield和Queues。幸好有wxPython wiki,它涵蓋了這些話題，因此如果你有興趣可以訪問(wèn)wiki的主頁(yè)，查看這些方法的使用。

您可能感興趣的文章: