R语言模拟工厂FIFO队列：实现工人完成任务后立即续接下一任务

我之前在Stack Overflow发布过《R: How to Simulate a Queue》，现在要模拟这类场景：

• 工厂有5名员工
• 箱子每日按泊松随机过程到达
• 员工遵循FIFO规则处理箱子：等待时间更长的优先处理
• 员工处理箱子时长服从指数分布，完成一个后立即处理下一个，从不空闲
• 若超过两名员工闲置，闲置最久的优先分配新箱子
• 工厂每日运营10小时，模拟10天的运行过程

我的现有实现代码如下，问题是当前每个工人每小时只能完成一个箱子，需要修改实现工人完成一个箱子后立即处理下一个的逻辑。

现有代码 #

步骤1：定义初始参数 #

library(data.table)

lambda_arrival <- 5  # 泊松到达率
lambda_service <- 3  # 指数服务率（每小时完成的箱子数）
hours_per_day <- 10
days <- 10
workers <- 5

results <- vector("list", days * hours_per_day)
completed_boxes <- vector("list", days * hours_per_day)

box_id_counter <- 1

步骤2：队列模拟函数 #

simulate_hour <- function(hour, day, queue, workers_status) {
    # 生成每小时到达的箱子数
    arrivals <- rpois(1, lambda_arrival)
    arrival_times <- rep(hour, arrivals)
    arrival_days <- rep(day, arrivals)
    arrival_ids <- seq(box_id_counter, box_id_counter + arrivals - 1)
    box_id_counter <<- box_id_counter + arrivals
    
    # 将新到达箱子加入队列
    if (arrivals > 0) {
        new_boxes <- data.table(box_id = arrival_ids, arrival_hour = arrival_times, arrival_day = arrival_days, arrival_id = arrival_ids, waiting_time = 0)
        queue <- rbind(queue, new_boxes)
    }
    
    # 处理箱子（当前逻辑：每个工人每小时仅处理1个）
    completed <- data.table(worker = character(0), box_id = numeric(0), completion_hour = numeric(0), completion_day = numeric(0), arrival_hour = numeric(0), arrival_day = numeric(0))
    for (i in 1:workers) {
        if (nrow(queue) > 0) {
            service_time <- rexp(1, lambda_service)
            queue[1, waiting_time := waiting_time + service_time]
            completed <- rbind(completed, data.table(worker = paste0("worker_", i), box_id = queue[1, box_id], completion_hour = hour, completion_day = day, arrival_hour = queue[1, arrival_hour], arrival_day = queue[1, arrival_day]))
            queue <- queue[-1]
        }
    }
    
    # 更新队列中剩余箱子的等待时间
    queue[, waiting_time := waiting_time + 1]
    
    # 存储每小时结果
    results[[hour + (day - 1) * hours_per_day]] <<- data.table(current_hour = hour, current_day = day, queue)
    completed_boxes[[hour + (day - 1) * hours_per_day]] <<- completed
    
    return(queue)
}

# 运行10天模拟
queue <- data.table(box_id = numeric(0), arrival_hour = numeric(0), arrival_day = numeric(0), arrival_id = numeric(0), waiting_time = 0)
for (day in 1:days) {
    for (hour in 1:hours_per_day) {
        queue <- simulate_hour(hour, day, queue, workers)
    }
}

步骤3：运行模拟 #

queue <- data.table(box_id = numeric(0), arrival_hour = numeric(0), arrival_day = numeric(0), arrival_id = numeric(0), waiting_time = 0)
for (day in 1:days) {
    for (hour in 1:hours_per_day) {
        queue <- simulate_hour(hour, day, queue, workers)
    }
}

print(results[[1]])
print(completed_boxes[[1]])

核心问题 #

当前代码中每个工人每小时只能处理1个箱子，无法实现“完成一个箱子后立即开始处理下一个”的逻辑，需要修改代码。

问题出在你当前的逻辑是按工人循环一次处理一个箱子，而不是跟踪每个工人的可用时间，让工人在一小时内尽可能处理多个箱子。核心修改点是：

1. 跟踪每个工人的可用时间戳（精确到小时内的小数，比如0.5小时）
2. 在每小时的时间窗口内，不断让空闲工人从队列取箱子处理，直到工人下一次完成时间超过当前小时结束，或者队列空了
3. 处理跨小时未完成的任务：工人未完成的任务要保留到下一小时继续处理

修改后的完整代码 #

library(data.table)

lambda_arrival <- 5  # 泊松到达率（每小时平均到达数）
lambda_service <- 3  # 指数服务率（每小时平均完成3个，即平均服务时间1/3小时）
hours_per_day <- 10
days <- 10
workers <- 5

# 初始化结果存储
results <- vector("list", days * hours_per_day)
completed_boxes <- vector("list", days * hours_per_day)
box_id_counter <- 1

# 模拟单小时的函数，核心是跟踪工人可用时间
simulate_hour <- function(hour, day, queue, workers_available) {
    # 计算当前小时的时间范围：start_time = day*10 + hour -10, end_time = start_time +1（比如第1天第1小时：0到1）
    start_time <- (day - 1)*hours_per_day + (hour - 1)
    end_time <- start_time + 1
    
    # 生成当前小时内的箱子到达事件（精确到小时内的时间点）
    arrivals <- rpois(1, lambda_arrival)
    if (arrivals > 0) {
        # 泊松过程的到达时间是均匀分布在小时内的
        arrival_times <- start_time + runif(arrivals)
        new_boxes <- data.table(
            box_id = seq(box_id_counter, box_id_counter + arrivals -1),
            arrival_time = arrival_times,
            arrival_day = day,
            arrival_hour = hour,
            waiting_time = 0
        )
        box_id_counter <<- box_id_counter + arrivals
        # 按到达时间排序，保证FIFO
        queue <- rbind(queue, new_boxes)[order(arrival_time)]
    }
    
    completed <- data.table(
        worker = character(0),
        box_id = numeric(0),
        completion_time = numeric(0),
        completion_day = numeric(0),
        completion_hour = numeric(0),
        arrival_time = numeric(0),
        arrival_day = numeric(0),
        arrival_hour = numeric(0),
        service_time = numeric(0)
    )
    
    # 循环处理：直到没有空闲工人，或队列空，或所有工人的可用时间都超过当前小时结束
    while(TRUE) {
        # 找到最早可用的工人和最早到达的箱子
        earliest_worker_idx <- which.min(workers_available)
        earliest_available_time <- workers_available[earliest_worker_idx]
        if (nrow(queue) == 0 || earliest_available_time >= end_time) {
            break
        }
        
        # 取队列最前面的箱子（FIFO）
        current_box <- queue[1]
        # 工人开始处理的时间：取工人可用时间和箱子到达时间的最大值（箱子没到的话工人要等）
        start_process_time <- max(earliest_available_time, current_box$arrival_time)
        # 生成服务时间
        service_time <- rexp(1, lambda_service)
        completion_time <- start_process_time + service_time
        
        # 如果完成时间在当前小时内，记录为已完成
        if (completion_time <= end_time) {
            completed <- rbind(completed, data.table(
                worker = paste0("worker_", earliest_worker_idx),
                box_id = current_box$box_id,
                completion_time = completion_time,
                completion_day = day,
                completion_hour = hour,
                arrival_time = current_box$arrival_time,
                arrival_day = current_box$arrival_day,
                arrival_hour = current_box$arrival_hour,
                service_time = service_time
            ))
            # 更新工人可用时间为完成时间
            workers_available[earliest_worker_idx] <- completion_time
            # 从队列移除已处理的箱子
            queue <- queue[-1]
        } else {
            # 任务跨小时，计算当前小时内的处理时长，更新箱子的等待时间
            queue[1, waiting_time := waiting_time + (end_time - start_process_time)]
            # 更新工人可用时间为当前小时结束时间，下一小时继续处理该箱子
            workers_available[earliest_worker_idx] <- end_time
            break
        }
    }
    
    # 更新队列中剩余箱子的等待时间（当前小时内从到达时间到小时结束的时长）
    queue[, waiting_time := waiting_time + (end_time - pmax(arrival_time, start_time))]
    
    # 存储当前小时的状态
    results[[hour + (day -1)*hours_per_day]] <<- data.table(
        current_day = day,
        current_hour = hour,
        queue = queue,
        workers_available = workers_available
    )
    completed_boxes[[hour + (day -1)*hours_per_day]] <<- completed
    
    return(list(queue = queue, workers_available = workers_available))
}

# 初始化：队列空，所有工人初始可用时间为0（第0小时开始）
initial_queue <- data.table(box_id = numeric(0), arrival_time = numeric(0), arrival_day = numeric(0), arrival_hour = numeric(0), waiting_time = 0)
workers_available <- rep(0, workers)

# 运行10天模拟
current_queue <- initial_queue
current_workers <- workers_available
for (day in 1:days) {
    for (hour in 1:hours_per_day) {
        sim_result <- simulate_hour(hour, day, current_queue, current_workers)
        current_queue <- sim_result$queue
        current_workers <- sim_result$workers_available
    }
}

# 查看第一天第一小时的结果
print(results[[1]])
print(completed_boxes[[1]])

修改说明 #

1. 时间戳精细化：把时间从“小时”粒度改成连续时间戳（比如第1天第1小时是0到1，第1天第2小时是1到2），精确跟踪工人可用时间和箱子到达时间。
2. 工人可用时间跟踪：用数组workers_available记录每个工人下一次可开始处理的时间，确保工人完成一个任务后立刻取下一个。
3. 循环处理逻辑：在每小时内循环分配空闲工人处理队列箱子，直到工人下一次完成时间超当前小时结束，或队列空。
4. 跨任务处理：若任务无法在当前小时完成，保留工人可用时间到下一小时，队列保留该箱子继续处理。
5. FIFO保障：队列始终按箱子到达时间排序，确保等待最久的箱子优先被处理。
6. 闲置工人优先级：通过which.min(workers_available)自动找到最早空闲的工人，实现“闲置最久的优先分配新箱子”规则。

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