本文實例講述了Python模擬簡單電梯調度算法。分享給大家供大家參考，具體如下：

經常在公司坐電梯，由于樓層較高，是雙聯裝的電梯，但是經常等電梯很久，經常有人罵寫電梯調度算法的。回來閑來無事，自己嘗試寫了一個簡單的。

場景很簡單，每一層電梯口只有一個按鈕，不區分上下，當有人按下這個鍵后，電梯會過來停在此層，這個人可以進去，并選擇自己想去的層。電梯的調度策略也很簡單，在一次向上的過程中，如果有人在下面按了鍵，電梯并不直接向下，而是運行到此次向上的最頂層，然后再下次向下運行的過程中去服務這個請求。

elevator.py

import timefrom myque import myqueclass elevator:  def __init__(self,layers):    self.building_layers = layers    self.direction = 'up'    self.cur_layer = 1    self.up_queue = myque()    self.down_queue = myque(True)    self.switcher = 'open'  def stop(self):    self.switcher='stop'  def push_button(self,layer,direction=None):    if self.cur_layer>layer:      self.down_queue.insert(layer)    elif self.cur_layer<layer:      self.up_queue.insert(layer)    else:      if self.direction=='up':        self.down_queue.insert(layer)      else:        self.up_queue.insert(layer)  def handle_queue(self,direction):    self.direction = direction    if direction == 'up':      inc = 1    else:      inc = -1    que = getattr(self , direction + '_queue')    while que.length():      while self.cur_layer != que.front():        print '/nelevator in ',self.cur_layer        time.sleep(1)        self.cur_layer += inc      print '/nelevator arrives at ',self.cur_layer      que.pop_front()  def run(self):    while self.switcher=='open':      if self.up_queue.empty() and self.down_queue.empty():        """elevator now is waiting, stop at a layer"""        time.sleep(1)        continue      """go up"""      self.handle_queue('up')      """go down"""      self.handle_queue('down')

myque.py

import threadingclass myque:  def __init__(self,reverse=False):    self.mode = reverse    self.buf = []    self.lock = threading.Lock()  def insert(self,object):    self.lock.acquire()    self.buf.append(object)    self.buf.sort(reverse = self.mode)    self.lock.release()  def front(self):    return self.buf[0]  def pop_front(self):    self.lock.acquire()    self.buf.pop(0)    self.lock.release()  def length(self):    self.lock.acquire()    size = len(self.buf)    self.lock.release()    return size  def empty(self):    self.lock.acquire()    size = len(self.buf)    self.lock.release()    return size==0

deploy.py

import threadingfrom elevator import elevatordef init_elevator(building_layers):  e = elevator(building_layers)  t = threading.Thread(target = e.run)  t.setDaemon(True)  t.start()  return (e,t)def main():  myelevator,ctl_thread = init_elevator(17)  while True:    str=raw_input("Input valid layer :")    try:      layer = int(str)    except Exception:      if str=='quit':        myelevator.stop()        ctl_thread.join()        break      else:        print 'invalid input',str        continue    if layer not in range(1,myelevator.building_layers+1):      continue    myelevator.push_button(layer)if __name__=='__main__':  main()