我想为电梯系统编写一个多线程程序。如果状态是running或wait状态，电梯系统只能移动。在系统中，可以以多线程方式添加指令（要转到的楼层）。为了管理指令队列并获得升降机的下一站，我编写了一个QueueManager类。我无法弄清楚如何实现QueueManager。问题是我无法理解编写QueueManager。特别是在弄清楚如何从队列中获取Lift的下一站。这是我在python3中对系统的实现：
"""Implementation of a lift System"""
from enum import Enum
from typing import TypeVar
from threading import Thread
import queue
LiftType = TypeVar('LiftType', bound='Lift')
QueueManagerType = TypeVar('QueueManagerType', bound='QueueManager')
SensorNameType = TypeVar('SensorNameType', bound='SensorName')
DirectionType = TypeVar('DirectionType', bound='DirectionType')
class SensorName(Enum):
weight = 'weight'
smoke = 'smoke'
class Direction(Enum):
up = 1
down = -1
class State(Enum):
running = 'running'
wait = 'waiting'
stop = 'stop'
class Lift:
def __init__(self, num_floors: int, queue_manager: QueueManagerType):
"""
Contains all the properties of lift
"""
self.num_floors = num_floors
self.curr_direction = Direction.up
self.state = State.running
self.curr_floor = 0
self.queue_manager = queue_manager
def move(self):
"""
Moves the lift according to the instruction
"""
if self.state in [State.running, State.wait]:
if self.queue_manager.has_instruction():
self.state = State.running
stop = self.queue_manager.next_stop(self.curr_direction, self.curr_floor)
if stop:
print(stop)
else:
if self.curr_direction == Direction.up:
self.curr_direction = Direction.down
stop = self.queue_manager.next_stop(self.curr_direction,
self.curr_floor)
print(stop)
else:
self.curr_direction = Direction.up
else:
self.state = State.wait
def add_instruction(self, floor, direction=None):
"""Adds the instruction to the queue"""
if direction is None:
if self.curr_floor > floor:
direction = Direction.down
else:
direction = Direction.up
Thread(target=self.queue_manager.add_instruction, args=(floor, direction)).start()
self.move()
class QueueManager:
def __init__(self):
self.instruction_queue = queue.Queue()
def add_instruction(self, floor: int, direction: DirectionType):
"""Add an instruction to the queue. Direction is used """
self.instruction_queue.put([floor, direction])
def next_stop(self, direction: int, curr_floor: int):
"""Tells which is the next stop based on the direction provided."""
pass
def has_instruction(self):
"""If there are any instructions for the lift"""
pass
if __name__ == '__main__':
# weight_sensor = WeightSensor(SensorName.weight)
instruction_queue = QueueManager()
lift_1 = Lift(21, instruction_queue)
lift_1.add_instruction(floor=0, direction=Direction.up)
lift_1.add_instruction(floor=2)
对上述程序的扩展是编写一个在后台运行的Sensor类，如果某个传感器被触发，则将电梯设置为停止状态。这也是我无法弄清楚的。
根据我在评论中得到的反馈。这是一个用于电梯系统的多线程解决方案。
"""Implementation of a lift System"""
from enum import Enum
from typing import TypeVar
from threading import Lock, Thread
from heapq import heappush, heappop
import time
import random
LiftType = TypeVar('LiftType', bound='Lift')
QueueManagerType = TypeVar('QueueManagerType', bound='QueueManager')
SensorNameType = TypeVar('SensorNameType', bound='SensorName')
DirectionType = TypeVar('DirectionType', bound='DirectionType')
class SensorName(Enum):
weight = 'weight'
smoke = 'smoke'
class Direction(Enum):
up = 1
down = -1
class State(Enum):
running = 'running'
wait = 'waiting'
stop = 'stop'
class Lift:
def __init__(self, num_floors: int):
"""
Contains all the properties of lift
:param sensors: A list of sensor objects
"""
self.num_floors = num_floors
self.curr_direction = Direction.up
self.state = State.running
self.curr_floor = 0
self.instruction_queue = {Direction.up: [], Direction.down: []}
self.queue_lock = Lock()
def move(self):
"""
Moves the lift according to the instruction
"""
if self.state in [State.running, State.wait]:
if len(self.instruction_queue[Direction.up]) or len(self.instruction_queue[Direction.down]):
self.state = State.running
try:
stop = heappop(self.instruction_queue[Direction.up])
except IndexError:
stop = -1
if stop > -1:
self.curr_floor = stop
print(stop, Direction.up)
else:
if self.curr_direction == Direction.up:
self.curr_direction = Direction.down
if len(self.instruction_queue[Direction.down]):
stop = heappop(self.instruction_queue[Direction.down])
self.curr_floor = stop
print(stop, Direction.down)
else:
self.state = State.wait
else:
self.curr_direction = Direction.up
self.move()
else:
self.state = State.wait
def add_instruction(self, floor, direction=None):
if direction == Direction.up:
time.sleep(random.uniform(0,1))
else:
time.sleep(random.uniform(0,4))
if direction is None:
if self.curr_floor > floor:
direction = Direction.down
else:
direction = Direction.up
with self.queue_lock:
heappush(self.instruction_queue[direction], floor)
self.move()
if __name__ == '__main__':
lift_1 = Lift(21)
lift_1.add_instruction(0, Direction.up)
Thread(target=lift_1.add_instruction, args=(2,)).start()
Thread(target=lift_1.add_instruction, args=(5,)).start()
Thread(target=lift_1.add_instruction, args=(2,)).start()
Thread(target=lift_1.add_instruction, args=(18,)).start()
Thread(target=lift_1.add_instruction, args=(12,)).start()
Thread(target=lift_1.add_instruction, args=(21,)).start()