Shell脚本如何实现任务队列

本文目录导读：

1. 基于文件的任务队列（最简单）
2. 基于命名管道的任务队列
3. 带有并发控制的队列
4. 带有资源限制的队列系统
5. 企业级队列管理（带状态记录）
6. 使用建议

Shell脚本实现任务队列有多种方式,从简单到复杂，我为你介绍几种常用方案：

基于文件的任务队列（最简单）

生产者-消费者模式

#!/bin/bash
QUEUE_DIR="/tmp/task_queue"
mkdir -p "$QUEUE_DIR"
# 生产者：添加任务
add_task() {
    local task_id=$(date +%s%N)
    local task_file="$QUEUE_DIR/$task_id"
    # 写入任务内容
    cat > "$task_file" << EOF
{
    "id": "$task_id",
    "command": "$1",
    "created_at": "$(date)"
}
EOF
    echo "Task $task_id added"
}
# 消费者：处理任务
process_tasks() {
    while true; do
        # 获取最早的任务文件
        local task_file=$(ls -t "$QUEUE_DIR" 2>/dev/null | tail -1)
        if [[ -n "$task_file" ]]; then
            local full_path="$QUEUE_DIR/$task_file"
            echo "Processing: $task_file"
            # 执行任务（使用source加载任务内容）
            source "$full_path"
            eval "$command"
            # 处理完成后删除任务文件
            rm "$full_path"
            echo "Completed: $task_file"
        fi
        sleep 1
    done
}
# 使用示例
add_task "echo 'Hello World'"
add_task "sleep 2; echo 'Task 2 done'"
add_task "date"
# 启动消费者（在后台运行）
process_tasks &

基于命名管道的任务队列

#!/bin/bash
PIPE="/tmp/task_pipe"
LOCK_FILE="/tmp/task_pipe.lock"
# 初始化管道
init_pipe() {
    if [[ ! -p "$PIPE" ]]; then
        mkfifo "$PIPE"
    fi
}
# 添加任务到队列
enqueue() {
    echo "$1" >> "$PIPE"
}
# 从队列取出任务
dequeue() {
    local task=""
    # 使用文件锁确保原子操作
    (
        flock -x 200
        # 读取一个任务
        read -r task < "$PIPE"
        if [[ -n "$task" ]]; then
            echo "$task"
        fi
    ) 200>"$LOCK_FILE"
}
# 工作进程
worker() {
    while true; do
        local task=$(dequeue)
        if [[ -n "$task" ]]; then
            echo "Worker $$ executing: $task"
            eval "$task"
            echo "Worker $$ completed: $task"
        fi
        sleep 0.1
    done
}
# 使用示例
init_pipe
# 添加任务
for i in {1..5}; do
    enqueue "echo 'Task $i: $(date)'"
done
# 启动2个工作进程
worker &
worker &
# 等待所有任务完成
wait

带有并发控制的队列

#!/bin/bash
MAX_CONCURRENT=3
QUEUE_FILE="/tmp/queue.txt"
RUNNING_FILE="/tmp/running.txt"
# 初始化队列
init_queue() {
    > "$QUEUE_FILE"
    > "$RUNNING_FILE"
}
# 添加任务
add_to_queue() {
    echo "$1" >> "$QUEUE_FILE"
}
# 获取任务计数
get_queue_count() {
    wc -l < "$QUEUE_FILE" 2>/dev/null || echo 0
}
# 获取运行中的任务数
get_running_count() {
    wc -l < "$RUNNING_FILE" 2>/dev/null || echo 0
}
# 执行任务
execute_task() {
    local task="$1"
    local task_id=$(date +%s%N)
    # 记录运行中的任务
    echo "$task_id:$task" >> "$RUNNING_FILE"
    echo "Starting: $task"
    # 执行任务
    eval "$task"
    # 从运行列表中移除
    sed -i "/^$task_id:/d" "$RUNNING_FILE"
    echo "Completed: $task"
}
# 调度器
scheduler() {
    while true; do
        local queue_count=$(get_queue_count)
        local running_count=$(get_running_count)
        # 检查是否有空闲槽位和待处理任务
        if [[ $running_count -lt $MAX_CONCURRENT && $queue_count -gt 0 ]]; then
            # 获取第一个任务
            local task=$(head -1 "$QUEUE_FILE")
            if [[ -n "$task" ]]; then
                # 从队列中移除
                sed -i '1d' "$QUEUE_FILE"
                # 在子进程中执行
                (execute_task "$task") &
            fi
        fi
        sleep 0.5
    done
}
# 使用示例
init_queue
# 添加一些耗时任务
add_to_queue "sleep 2; echo 'Task 1 done'"
add_to_queue "sleep 3; echo 'Task 2 done'"
add_to_queue "sleep 1; echo 'Task 3 done'"
add_to_queue "sleep 4; echo 'Task 4 done'"
add_to_queue "sleep 2; echo 'Task 5 done'"
# 启动调度器
scheduler &
# 等待完成
sleep 10
echo "All tasks dispatched"

带有资源限制的队列系统

#!/bin/bash
# 任务队列类
declare -A TASK_QUEUE
TASK_COUNTER=0
MAX_TASKS=10
MAX_MEMORY=500  # MB
# 检查系统资源
check_resources() {
    local memory_usage=$(free -m | awk 'NR==2{print $3}')
    local cpu_usage=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
    if [[ $memory_usage -gt $MAX_MEMORY ]]; then
        return 1  # 内存不足
    fi
    if [[ $(echo "$cpu_usage > 80" | bc) -eq 1 ]]; then
        return 1  # CPU负载过高
    fi
    return 0
}
# 添加任务
add_task() {
    local priority=${2:-5}  # 默认优先级5
    local task_id=$((++TASK_COUNTER))
    TASK_QUEUE[$task_id]="$1:$priority"
    # 按优先级排序
    sort_tasks
}
# 排序任务（按优先级）
sort_tasks() {
    # 创建临时排序文件
    local temp_file=$(mktemp)
    for id in "${!TASK_QUEUE[@]}"; do
        IFS=':' read -r task priority <<< "${TASK_QUEUE[$id]}"
        echo "$priority $id $task" >> "$temp_file"
    done
    sort -n "$temp_file" | while read line; do
        echo "$line"
    done
    rm "$temp_file"
}
# 获取最高优先级任务
get_next_task() {
    local highest_priority=999
    local next_task_id=""
    local next_command=""
    for id in "${!TASK_QUEUE[@]}"; do
        IFS=':' read -r task priority <<< "${TASK_QUEUE[$id]}"
        if [[ $priority -lt $highest_priority ]]; then
            highest_priority=$priority
            next_task_id=$id
            next_command=$task
        fi
    done
    if [[ -n "$next_task_id" ]]; then
        echo "$next_task_id:$next_command"
    fi
}
# 执行任务
run_task() {
    local command="$1"
    if check_resources; then
        echo "Executing: $command"
        eval "$command"
        return $?
    else
        echo "Resource limit reached, waiting..."
        sleep 2
        run_task "$command"  # 重试
    fi
}
# 任务管理器
task_manager() {
    while true; do
        local next_task=$(get_next_task)
        if [[ -n "$next_task" ]]; then
            local task_id="${next_task%%:*}"
            local command="${next_task#*:}"
            # 删除任务
            unset TASK_QUEUE[$task_id]
            # 执行任务
            run_task "$command"
            echo "Task $task_id completed"
        fi
        sleep 1
    done
}
# 使用示例
add_task "echo 'High priority task'" 1
add_task "sleep 2; echo 'Normal task'"
add_task "sleep 1; echo 'Low priority task'" 10
# 启动任务管理器
task_manager &

企业级队列管理（带状态记录）

#!/bin/bash
# 队列数据库（使用文件模拟）
QUEUE_DB="/tmp/queue.db"
STATUS_DB="/tmp/queue_status.db"
# 初始化数据库
init_db() {
    echo "task_id|status|command|created_at|completed_at" > "$QUEUE_DB"
    echo "idle" > "$STATUS_DB"
}
# 添加任务到数据库
db_add_task() {
    local task_id=$(uuidgen 2>/dev/null || date +%s%N)
    local command="$1"
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    echo "$task_id|pending|$command|$timestamp|" >> "$QUEUE_DB"
    echo "$task_id"
}
# 更新任务状态
db_update_status() {
    local task_id="$1"
    local new_status="$2"
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    # 使用临时文件更新
    local temp_file=$(mktemp)
    while IFS='|' read -r id status command created completed; do
        if [[ "$id" == "$task_id" ]]; then
            if [[ "$new_status" == "completed" || "$new_status" == "failed" ]]; then
                echo "$id|$new_status|$command|$created|$timestamp"
            else
                echo "$id|$new_status|$command|$created|$completed"
            fi
        else
            echo "$id|$status|$command|$created|$completed"
        fi
    done < "$QUEUE_DB" > "$temp_file"
    mv "$temp_file" "$QUEUE_DB"
}
# 获取待处理任务
get_pending_tasks() {
    awk -F'|' '$2 == "pending" {print $0}' "$QUEUE_DB"
}
# 任务执行器
task_executor() {
    local worker_id=$1
    echo "Worker $worker_id started"
    while true; do
        # 获取当前系统状态
        local queue_status=$(cat "$STATUS_DB")
        if [[ "$queue_status" == "running" ]]; then
            # 获取待处理任务
            local pending_tasks=$(get_pending_tasks)
            if [[ -n "$pending_tasks" ]]; then
                # 取第一个任务
                local task=$(echo "$pending_tasks" | head -1)
                local task_id=$(echo "$task" | cut -d'|' -f1)
                local command=$(echo "$task" | cut -d'|' -f3)
                # 更新状态为运行中
                db_update_status "$task_id" "running"
                echo "Worker $worker_id executing: $command"
                # 执行任务
                if eval "$command"; then
                    db_update_status "$task_id" "completed"
                    echo "Worker $worker_id completed task $task_id"
                else
                    db_update_status "$task_id" "failed"
                    echo "Worker $worker_id failed task $task_id"
                fi
            fi
        fi
        sleep 2
    done
}
# 队列控制器
queue_controller() {
    echo "Queue controller started"
    # 添加示例任务
    for i in {1..5}; do
        db_add_task "echo 'Task $i: $(date)'; sleep 2"
    done
    # 设置队列为运行状态
    echo "running" > "$STATUS_DB"
    # 启动3个工作进程
    for i in {1..3}; do
        task_executor "$i" &
    done
    # 监控队列状态
    while true; do
        local pending=$(get_pending_tasks | wc -l)
        local total=$(wc -l < "$QUEUE_DB")
        echo "Queue status: $pending pending / $((total - 1)) total"
        if [[ $pending -eq 0 ]]; then
            echo "All tasks completed"
            echo "idle" > "$STATUS_DB"
            break
        fi
        sleep 5
    done
}
# 使用示例
init_db
queue_controller

使用建议

1. 简单场景：使用基于文件的队列（方法1）
2. 并发控制：使用命名管道方案（方法2）
3. 资源限制：使用带有资源控制的队列（方法3-4）
4. 企业级应用：使用带数据库记录的方案（方法5）

这些方案可以根据你的具体需求进行修改和扩展,需要我详细解释某个具体实现吗？