车联网场景下基于Python的时延敏感任务调度建模问题

核心思路：用模拟而非真实阻塞 #

别用time.sleep()，它会阻塞主线程，完全没法模拟并行调度场景。我们需要用事件驱动模拟或手动维护时间线，结合多线程/多进程实现并行逻辑。

具体实现方案 #

1. 封装数据结构 #

先把任务和边缘服务器封装成类，方便状态管理：

import pandas as pd
import math
from threading import Thread, Lock
from queue import Queue
import time

class Task:
    def __init__(self, task_id, file_size, length):
        self.task_id = task_id
        self.file_size = file_size
        self.length = length
        self.assigned_server = None
        self.completion_time = None
        self.migration_time = 0

class EdgeServer:
    def __init__(self, server_id, x, y, cp, dtr):
        self.server_id = server_id
        self.x = x
        self.y = y
        self.cp = cp  # 计算能力
        self.dtr = dtr  # 数据传输速率
        self.state = "idle"
        self.lock = Lock()  # 线程安全锁，防止状态冲突

    def computation_time(self, task_length):
        # 示例计算耗时公式，可根据需求调整
        return task_length / self.cp

    def migration_time(self, file_size):
        # 示例迁移耗时公式，可加入距离权重
        return file_size / self.dtr

    def distance_to(self, other_server):
        # 计算欧氏距离
        return math.hypot(self.x - other_server.x, self.y - other_server.y)

2. 并行调度逻辑 #

用线程池+任务队列处理并行的任务分配与执行，避免手动管理多线程的复杂度：

class Scheduler:
    def __init__(self, servers):
        self.servers = servers
        self.task_queue = Queue()
        self.threads = []
        # 启动多个调度线程，模拟并行处理
        for _ in range(5):
            t = Thread(target=self._process_tasks)
            t.daemon = True
            t.start()
            self.threads.append(t)

    def _find_nearest_idle_server(self, target_server):
        # 按距离排序，找最近的空闲服务器
        sorted_servers = sorted(self.servers.values(), key=lambda s: s.distance_to(target_server))
        for server in sorted_servers:
            with server.lock:
                if server.state == "idle":
                    return server
        return None

    def _process_tasks(self):
        while True:
            task = self.task_queue.get()
            if task is None:
                break
            # 随机选初始服务器
            initial_server = list(self.servers.values())[task.task_id % len(self.servers)]
            
            with initial_server.lock:
                if initial_server.state == "idle":
                    assigned_server = initial_server
                    assigned_server.state = "busy"
                else:
                    # 找最近空闲服务器
                    assigned_server = self._find_nearest_idle_server(initial_server)
                    if assigned_server:
                        task.migration_time = assigned_server.migration_time(task.file_size)
                        assigned_server.state = "busy"
                    else:
                        # 无空闲服务器，重新入队等待
                        time.sleep(0.1)
                        self.task_queue.put(task)
                        continue
            
            # 模拟计算耗时（真实场景可替换为实际计算逻辑）
            compute_time = assigned_server.computation_time(task.length)
            time.sleep(compute_time)
            
            # 任务完成，释放服务器
            with assigned_server.lock:
                assigned_server.state = "idle"
            task.completion_time = compute_time + task.migration_time
            self.task_queue.task_done()

    def add_task(self, task):
        self.task_queue.put(task)

3. 加载数据并运行模拟 #

# 加载任务DataFrame
tasks_df = pd.DataFrame({
    "FILE_SIZE": [100, 200, 150],
    "LENGTH": [500, 800, 600]
})

# 加载边缘服务器字典
servers = {
    1: EdgeServer(1, 0, 0, 100, 50),
    2: EdgeServer(2, 10, 0, 150, 60),
    3: EdgeServer(3, 0, 10, 120, 55)
}

# 初始化调度器
scheduler = Scheduler(servers)

# 创建任务并加入队列
task_list = []
for idx, row in tasks_df.iterrows():
    task = Task(idx, row["FILE_SIZE"], row["LENGTH"])
    task_list.append(task)
    scheduler.add_task(task)

# 等待所有任务完成
scheduler.task_queue.join()

# 统计输出结果
for task in task_list:
    print(f"任务{task.task_id}: 总耗时={task.completion_time:.2f}, 迁移耗时={task.migration_time:.2f}")

关键注意事项 #

• 线程安全：所有服务器状态修改必须加Lock()，避免多线程同时修改导致状态混乱。
• 模拟效率：如果是大规模模拟，推荐用SimPy离散事件模拟框架替代time.sleep()，它可以直接推进时间线，不用真实等待，效率更高。
• 多进程vs多线程：如果computation_time()是CPU密集型任务，用multiprocessing替代threading；如果是IO密集型（模拟网络传输、等待），用线程足够。

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