如何设计多线程电梯系统并实现QueueManager类处理并发？

Hey there! Let’s work through your QueueManager implementation—this is a super common concurrency problem, so I’ll break down the key pieces you’re missing and give you a working example to build from.

First, let’s recap the core requirements to make sure we’re aligned:

• The elevator system only accepts new floor requests when in RUNNING or WAIT state
• Multiple threads need to safely add target floor instructions
• The QueueManager needs to track pending requests and determine the elevator’s next stop

Key Design Principles for QueueManager #

Before diving into code, let’s cover the critical concepts you need to handle:

1. Thread Safety: Any data shared between threads (like the request queue, elevator state, or pending floors) must be protected to avoid race conditions.
2. State Validation: Reject any new requests if the elevator isn’t in an operational state (RUNNING/WAIT).
3. Request Deduplication: Avoid processing duplicate requests for the same floor (no need to stop twice for the same call!).
4. Clear Next-Stop Logic: Decide the next floor based on elevator position and pending requests (we’ll start with a simple FIFO approach, then touch on real-world elevator algorithms).

Example Implementation (Java) #

Let’s use Java since it’s widely used for multi-threaded systems, but I’ll also note how to adapt this to Python later.

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.HashSet;
import java.util.Set;

public class QueueManager {
    // Thread-safe queue for floor requests
    private final ConcurrentLinkedQueue<Integer> requestQueue = new ConcurrentLinkedQueue<>();
    // Track pending floors to avoid duplicates
    private final Set<Integer> pendingFloors = new HashSet<>();
    // Volatile ensures thread visibility of state/floor changes
    private volatile ElevatorState currentState;
    private volatile int currentFloor;

    public enum ElevatorState {
        RUNNING, WAIT, STOPPED // STOPPED = non-operational state
    }

    public QueueManager(int initialFloor) {
        this.currentFloor = initialFloor;
        this.currentState = ElevatorState.WAIT; // Start in idle state
    }

    // Thread-safe method to add target floors (only allowed in RUNNING/WAIT)
    public boolean addTargetFloor(int floor) {
        // First check if the elevator is operational
        if (currentState != ElevatorState.RUNNING && currentState != ElevatorState.WAIT) {
            return false; // Request rejected
        }

        // Protect duplicate check with synchronized block (HashSet isn't thread-safe)
        synchronized (pendingFloors) {
            if (pendingFloors.add(floor)) {
                requestQueue.offer(floor);
                return true; // Request added successfully
            }
        }
        return false; // Floor already pending
    }

    // Get the next stop and update state/queue
    public Integer getNextStation() {
        // Don't process requests if elevator is stopped
        if (currentState != ElevatorState.RUNNING && currentState != ElevatorState.WAIT) {
            return null;
        }

        Integer nextFloor = requestQueue.poll();
        if (nextFloor != null) {
            // Remove from pending set once we start processing
            synchronized (pendingFloors) {
                pendingFloors.remove(nextFloor);
            }
            // Switch to running state if we have a request
            currentState = ElevatorState.RUNNING;
        } else {
            // No pending requests: switch to wait state, stay at current floor
            currentState = ElevatorState.WAIT;
            nextFloor = currentFloor;
        }
        return nextFloor;
    }

    // Update current floor once elevator arrives
    public void updateCurrentFloor(int floor) {
        this.currentFloor = floor;
    }

    // Modify elevator state (e.g., emergency stop)
    public void setState(ElevatorState state) {
        this.currentState = state;
    }

    // Getters for external components (elevator thread)
    public ElevatorState getCurrentState() {
        return currentState;
    }

    public int getCurrentFloor() {
        return currentFloor;
    }
}

Key Details Explained #

Let’s break down the parts you might have been stuck on:

• Thread Safety:
  • ConcurrentLinkedQueue is used for the request queue because it’s inherently thread-safe for add/remove operations.
  • The pendingFloors HashSet isn’t thread-safe, so we wrap its operations in a synchronized block to prevent race conditions when checking for duplicates.
  • volatile modifiers on currentState and currentFloor ensure that changes made by one thread are immediately visible to others (no stale reads).
• State Control: The addTargetFloor method first validates the elevator’s state—if it’s stopped, requests are rejected immediately.
• Next-Stop Logic: This example uses a simple FIFO approach, but you can extend it to real-world elevator algorithms like SCAN (move in one direction until no more requests, then reverse) or LOOK (similar to SCAN but stops at the last request in the current direction). For example, you could iterate the queue to find the closest floor in the elevator’s current direction instead of just taking the first item.

Python Adaptation #

If you’re working in Python, here’s how to translate the logic using Python’s built-in thread-safe tools:

import queue
import threading
from enum import Enum

class ElevatorState(Enum):
    RUNNING = 1
    WAIT = 2
    STOPPED = 3

class QueueManager:
    def __init__(self, initial_floor):
        self.request_queue = queue.Queue()
        self.pending_floors = set()
        self.pending_lock = threading.Lock()  # Protects pending_floors
        self.current_state = ElevatorState.WAIT
        self.state_lock = threading.Lock()    # Protects current_state
        self.current_floor = initial_floor

    def add_target_floor(self, floor):
        # Check operational state first
        with self.state_lock:
            if self.current_state not in (ElevatorState.RUNNING, ElevatorState.WAIT):
                return False
        
        # Prevent duplicate requests
        with self.pending_lock:
            if floor not in self.pending_floors:
                self.pending_floors.add(floor)
                self.request_queue.put(floor)
                return True
        return False

    def get_next_station(self):
        # Validate state before processing
        with self.state_lock:
            if self.current_state not in (ElevatorState.RUNNING, ElevatorState.WAIT):
                return None
        
        try:
            next_floor = self.request_queue.get_nowait()
            with self.pending_lock:
                self.pending_floors.remove(next_floor)
            with self.state_lock:
                self.current_state = ElevatorState.RUNNING
            return next_floor
        except queue.Empty:
            # No requests: switch to wait state
            with self.state_lock:
                self.current_state = ElevatorState.WAIT
            return self.current_floor

    def update_current_floor(self, floor):
        self.current_floor = floor

    def set_state(self, state):
        with self.state_lock:
            self.current_state = state

Next Steps to Expand #

• Add direction tracking (up/down) to implement smarter scheduling algorithms.
• Add timeout handling for requests if needed.
• Integrate with your elevator thread to consume getNextStation() updates and move the elevator accordingly.

内容的提问来源于stack exchange，提问作者Mudits