Shell脚本如何实现消息队列

本文目录导读：

1. 使用文件系统作为消息队列（最基础）
2. 使用FIFO/命名管道
3. 基于Redis的简单队列
4. 使用内存磁盘或tmpfs
5. 基于netcat的简单消息队列
6. 完整的消息队列系统
7. 使用建议

我来详细介绍几种在Shell脚本中实现消息队列的方法：

使用文件系统作为消息队列（最基础）

生产者脚本 (producer.sh)

#!/bin/bash
QUEUE_DIR="./message_queue"
mkdir -p "$QUEUE_DIR"
# 发送消息
send_message() {
    local message="$1"
    local timestamp=$(date +%s%N)
    local filename="${QUEUE_DIR}/msg_${timestamp}_${RANDOM}"
    echo "$message" > "$filename"
    echo "发送消息: $message"
}
# 示例
send_message "用户登录"
send_message "订单创建"

消费者脚本 (consumer.sh)

#!/bin/bash
QUEUE_DIR="./message_queue"
# 处理消息
process_message() {
    local file="$1"
    local content=$(cat "$file")
    echo "处理消息: $content"
    rm "$file"  # 处理完后删除
}
# 监听队列
while true; do
    for file in "$QUEUE_DIR"/msg_*; do
        if [ -f "$file" ]; then
            process_message "$file"
        fi
    done
    sleep 1
done

使用FIFO/命名管道

创建和管理管道

#!/bin/bash
PIPE_FILE="/tmp/msg_queue.pipe"
# 创建命名管道
if [ ! -p "$PIPE_FILE" ]; then
    mkfifo "$PIPE_FILE"
fi
# 生产者
producer() {
    local message="$1"
    echo "$message" > "$PIPE_FILE"
    echo "发送: $message"
}
# 消费者
consumer() {
    while true; do
        if read line < "$PIPE_FILE"; then
            echo "收到: $line"
            # 处理消息
            case "$line" in
                "SHUTDOWN") exit 0 ;;
                *) echo "处理消息: $line" ;;
            esac
        fi
    done
}
# 使用示例
case "$1" in
    "send") producer "$2" ;;
    "receive") consumer ;;
esac

基于Redis的简单队列

需要先安装redis-cli

#!/bin/bash
REDIS_CLI="redis-cli"
QUEUE_NAME="shell_queue"
# 入队
enqueue() {
    $REDIS_CLI rpush "$QUEUE_NAME" "$1"
    echo "入队: $1"
}
# 出队
dequeue() {
    local message=$($REDIS_CLI lpop "$QUEUE_NAME")
    if [ -n "$message" ]; then
        echo "出队: $message"
        # 处理消息
        process "$message"
    fi
}
# 消息处理
process() {
    echo "处理消息: $1"
    # 业务逻辑
}
# 批量处理
batch_process() {
    while true; do
        dequeue
        sleep 0.1
    done
}

使用内存磁盘或tmpfs

#!/bin/bash
# 在内存中创建队列目录
QUEUE_BASE="/dev/shm/msg_queue"
mkdir -p "$QUEUE_BASE"
# 带锁的消息队列实现
lock_file() {
    local file="$1"
    local lock="${file}.lock"
    while ! mkdir "$lock" 2>/dev/null; do
        sleep 0.01
    done
}
unlock_file() {
    local lock="${1}.lock"
    rm -rf "$lock"
}
# 消息入队
enqueue_msg() {
    local msg="$1"
    local priority="${2:-0}"
    local timestamp=$(date +%s)
    local filename="${QUEUE_BASE}/${priority}_${timestamp}_$$"
    lock_file "$filename"
    echo "$msg" > "$filename"
    unlock_file "$filename"
}
# 消息出队
dequeue_msg() {
    # 按优先级排序处理
    for file in $(ls -r "$QUEUE_BASE" 2>/dev/null | sort -t_ -k1); do
        local fullpath="${QUEUE_BASE}/${file}"
        if [ -f "$fullpath" ]; then
            lock_file "$fullpath"
            local content=$(cat "$fullpath")
            rm "$fullpath"
            unlock_file "$fullpath"
            echo "$content"
            return 0
        fi
    done
    return 1
}

基于netcat的简单消息队列

消息队列服务端 (server.sh)

#!/bin/bash
QUEUE_FILE="/tmp/messages.dat"
PORT=5555
handle_message() {
    local message="$1"
    echo "$(date) - 收到: $message" >> "$QUEUE_FILE"
    echo "ACK"
}
# 启动服务
echo "消息队列服务启动在端口 $PORT"
while true; do
    message=$(nc -l -p "$PORT")
    [ -n "$message" ] && handle_message "$message"
done

客户端 (client.sh)

#!/bin/bash
send_message() {
    echo "$1" | nc localhost 5555
    echo "消息已发送: $1"
}
# 使用示例
send_message "Hello World"
send_message "这是一条测试消息"

完整的消息队列系统

下面是一个更完整、实用的实现：

#!/bin/bash
# 配置
QUEUE_DIR="${HOME}/.message_queue"
MAX_RETRIES=3
RETRY_DELAY=5
# 初始化
init_queue() {
    mkdir -p "${QUEUE_DIR}/{new,processing,failed,backup}"
    echo "队列初始化完成"
}
# 发送消息
send_msg() {
    local queue="$1"
    local message="$2"
    local timestamp=$(date +%Y%m%d_%H%M%S_%N)
    local filename="${QUEUE_DIR}/new/${queue}_${timestamp}_$$.msg"
    echo "$message" > "$filename"
    echo "消息已入队: $filename"
}
# 处理消息
process_msg() {
    local file="$1"
    local content=$(cat "$file")
    local basename=$(basename "$file")
    local queue=$(echo "$basename" | cut -d_ -f1)
    echo "处理队列 $queue 的消息: $content"
    # 模拟处理
    case "$queue" in
        "email") sleep 1 ;;
        "sms") sleep 0.5 ;;
        "log") sleep 0.1 ;;
        *) sleep 0.3 ;;
    esac
    return 0  # 成功
}
# 消费消息
consume() {
    local queue="${1:-*}"
    for file in "${QUEUE_DIR}/new/${queue}_"*.msg; do
        [ ! -f "$file" ] && continue
        local basename=$(basename "$file")
        local target="${QUEUE_DIR}/processing/${basename}"
        # 移动到处理中目录
        mv "$file" "$target"
        # 处理消息
        if process_msg "$target"; then
            # 成功：移动到备份
            mv "$target" "${QUEUE_DIR}/backup/"
            echo "消息处理成功: $basename"
        else
            # 失败：重试机制
            local retry_count=$(cat "${target}.retry" 2>/dev/null || echo "0")
            retry_count=$((retry_count + 1))
            if [ "$retry_count" -le "$MAX_RETRIES" ]; then
                echo "$retry_count" > "${target}.retry"
                sleep "$RETRY_DELAY"
                mv "$target" "${QUEUE_DIR}/new/"
                echo "消息将重试: $basename (第${retry_count}次)"
            else
                mv "$target" "${QUEUE_DIR}/failed/"
                echo "消息处理失败: $basename"
            fi
        fi
    done
}
# 监控队列状态
queue_status() {
    echo "======= 队列状态 ======="
    echo "新消息: $(ls ${QUEUE_DIR}/new/ 2>/dev/null | wc -l)"
    echo "处理中: $(ls ${QUEUE_DIR}/processing/ 2>/dev/null | wc -l)"
    echo "失败: $(ls ${QUEUE_DIR}/failed/ 2>/dev/null | wc -l)"
    echo "备份: $(ls ${QUEUE_DIR}/backup/ 2>/dev/null | wc -l)"
}
# 主函数
main() {
    case "$1" in
        "init") init_queue ;;
        "send") send_msg "$2" "$3" ;;
        "consume") consume "$2" ;;
        "status") queue_status ;;
        "daemon")
            while true; do
                consume "$2"
                sleep 1
            done
            ;;
        *)
            echo "用法: $0 {init|send|consume|status|daemon} [参数]"
            echo "示例:"
            echo "  $0 init                     - 初始化队列"
            echo "  $0 send email 'Hello'       - 发送消息"
            echo "  $0 consume                  - 消费一次"
            echo "  $0 daemon email             - 守护进程模式"
            echo "  $0 status                   - 查看状态"
            ;;
    esac
}
main "$@"

使用建议

1. 简单场景：使用文件系统实现，易于理解和调试
2. 性能要求：使用内存磁盘或命名管道
3. 分布式系统：结合Redis或消息中间件
4. 生产环境：使用专业的消息队列系统（RabbitMQ, Kafka等）

这个实现包含了消息持久化、重试机制、失败处理等特性,适合大多数Shell脚本中的消息队列需求。