Mi&Bee Blog
English

Bash高级编程技巧——用生产者消费者模式处理消息队列积压

Linux Bash, Shell, Linux, Scripting 3373 字 7 分钟阅读
🔊

前言

在消息系统运维中，队列积压是一个常见且棘手的问题。遭遇恶意攻击或系统异常时，投递队列可能出现大量消息积压，严重影响系统正常运转；传统的单进程处理方式效率低下，处理不及时还会让问题进一步恶化。

本文介绍一套基于生产者消费者模式的 Bash 方案，通过多进程并发、FIFO 管道通信、awk 解析和文件描述符信号量，实现消息队列积压的快速批量处理。该方案已在多个生产环境验证，处理效率较单进程提升数倍。

问题场景

通用化问题描述

消息投递系统面临的主要挑战包括：

  1. 恶意攻击导致的队列积压：攻击者短时间内发送大量垃圾邮件，导致投递队列文件激增
  2. 系统故障引起的资源争用：反垃圾引擎或其他服务异常，占用大量CPU和IO资源
  3. 突发业务高峰：营销活动或节日祝福等场景下，正常邮件投递量急剧增加
  4. 长期积累的性能瓶颈：随着时间推移，临时文件增多，系统性能逐渐下降

典型症状表现

当消息队列出现问题时，通常会有以下症状：

  • 队列文件数量超过正常阈值（如数千甚至上万）
  • 投递延迟显著增加，用户收件时间大幅延长
  • 系统负载持续高位，影响正常业务处理
  • 磁盘空间快速消耗，存在系统崩溃风险

设计思路

多进程管道架构

本次方案采用生产者消费者模式，通过多进程并发处理来提升效率。核心设计思路如下：

mermaid
flowchart TD
    classDef primary fill:#e3f2fd,stroke:#1976d2,stroke-width:2px
    classDef storage fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    classDef process fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    classDef network fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    
    A@{ shape: process, label: "日志扫描/文件扫描" } --> B@{ shape: cyl, label: "消息队列" }
    B --> C@{ shape: process, label: "生产者进程" }
    C --> D@{ shape: hex, label: "管道通信" }
    D --> E@{ shape: process, label: "消费者进程" }
    E --> F@{ shape: process, label: "消息移动处理" }
    F --> G@{ shape: rounded, label: "结果汇总" }
    
    class A C E F process
    class B storage
    class D network
    class G primary
    
    subgraph "生产者端"
        A
        B
        C
    end
    
    subgraph "消费者端"
        D
        E
        F
        G
    end

关键设计要点

  1. 分离生产与消费：将扫描任务和移动任务分离，避免互相阻塞
  2. 管道通信：使用FIFO管道实现进程间高效通信
  3. 并发控制：通过信号量机制控制并发进程数量
  4. 错误处理：完善的日志记录和异常处理机制
  5. 资源管理：自动清理临时文件，避免资源泄露

性能优势分析

与传统的单进程处理相比，本方案具有以下优势：

  1. 并行处理：多个消费者进程同时工作，线性提升处理速度
  2. 负载均衡：自动将任务分配到不同进程，避免单点瓶颈
  3. IO优化：批量处理文件系统操作，减少IO开销
  4. 内存友好：管道流式处理，避免大量数据驻留内存

核心实现

1. 队列扫描模块

队列扫描是整个处理流程的第一步，负责识别需要处理的目标文件。这里我们提供两种扫描方式：

基于上下文文件的扫描

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

#!/usr/bin/env bash

# 扫描消息上下文文件，匹配目标邮件
scan_context_files() {
    local pattern="$1"
    local context_dir="/opt/app/var/mda/dacontext/"
    local backup_dir="/opt/app/backup/$(date +%F)/"
    
    # 创建备份目录
    mkdir -p "${backup_dir}"{msg,dacontext}
    
    # 查找并扫描上下文文件
    find "${context_dir}" -type f | while read line; do
        if [[ -n $(grep -a "${pattern}" "$line") ]]; then
            echo "$(basename "$line")"
        fi
    done
}

基于日志文件的扫描

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

#!/usr/bin/env bash

# 扫描投递日志，提取目标消息ID
scan_log_files() {
    local log_dir="/opt/app/logs/rcptstat/"
    local pattern="$1"
    local today=$(date +%F)
    local yesterday=$(date +%F -d "yesterday")
    
    # 处理今天和昨天的日志文件
    for day in "${today}" "${yesterday}"; do
        local day_log="${log_dir}/da${day//-/_}"
        
        for ((i=0; i<24; i++)); do
            local log_file="${day_log}/da.127.0.0.1.$(date +%H -d "-$i hour")_00_00.log"
            
            if [[ -f "${log_file}" ]]; then
                process_log_file "${log_file}" "${pattern}"
            fi
        done
    done
}

2. 多进程并发架构

实现多进程并发的核心是管道和信号量机制：

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

#!/usr/bin/env bash

# 设置并发进程数
PROCESS_NUMS=10

# 创建用于进程控制的管道
TMP_FIFO_FOR_PROCESS="/tmp/$$_06.fifo" && \
    mkfifo "${TMP_FIFO_FOR_PROCESS}" && \
    exec 6<>"${TMP_FIFO_FOR_PROCESS}" && \
    rm "${TMP_FIFO_FOR_PROCESS}"

# 初始化信号量
for ((i=0; i<${PROCESS_NUMS}; i++)); do
    echo
done >&6

# 创建用于任务传递的管道
TMP_FIFO_FOR_TID="/tmp/$$_07.fifo" && \
    mkfifo "${TMP_FIFO_FOR_TID}" && \
    exec 7<>"${TMP_FIFO_FOR_TID}" && \
    rm "${TMP_FIFO_FOR_TID}"

3. 生产者进程实现

生产者进程负责扫描文件并将任务发送到管道：

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#!/usr/bin/env bash

# 生产者：扫描上下文文件并发送任务
producer_context_scan() {
    local pattern="$1"
    local context_dir="/opt/app/var/mda/dacontext/"
    
    # 启动扫描进程
    find "${context_dir}" -type f | while read line; do
        if [[ -n $(grep -a "${pattern}" "$line") ]]; then
            echo "$(basename "$line")" >&7 2>/dev/null
        fi
    done
    
    # 发送结束标记
    echo "done" >&7
}

# 生产者：扫描日志文件并发送任务  
producer_log_scan() {
    for file in "$@"; do
        if [[ ! -f "$file" ]]; then
            continue
        fi
        
        {
            # 处理日志文件，提取消息ID
            awk -F"[][]" '
            $0 ~ /\[cmd:remote,tid/ {
                tid=gettid($2)
                result=getresult($2)
                if(result=="Normal-Spam"){
                    print tid
                }
            }
            
            function gettid(s){
                match(s,/,tid:.*[0-9],from:/)
                start=RSTART + 5
                end=RLENGTH - 11
                return substr(s,start,end)
            }
            
            function getresult(s){
                match(s,/,commresult:.*,commrefid:/)
                start=RSTART + 12
                end=RLENGTH - 23
                return substr(s,start,end)
            }' "$file" | while read line; do
                echo "${line}" >&7 2>/dev/null
            done
        } &
    done
    
    # 发送结束标记
    echo "done" >&7
}

4. 消费者进程实现

消费者进程从管道获取任务并执行具体的移动操作：

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

#!/usr/bin/env bash

# 消费者：处理消息移动任务
consumer_move_messages() {
    local backup_dir="/opt/app/backup/$(date +%F)/"
    local msg_dir="/opt/app/var/mda/msg/"
    local context_dir="/opt/app/var/mda/dacontext/"
    
    # 创建备份目录
    mkdir -p "${backup_dir}"{msg,dacontext}
    
    while true; do
        if read tid <&7; then
            [[ "${tid}" == 'done' ]] && wait && break
            
            # 获取信号量
            read -u6
            
            # 执行移动操作
            {
                # 移动消息文件
                find "${msg_dir}" -name "${tid}" -exec mv {} "${backup_dir}/" \;
                
                # 移动上下文文件
                find "${context_dir}" -name "${tid}" -exec mv {} "${backup_dir}/dacontext/" \;
                
                # 记录日志
                echo "$(date '+%F %H:%M') - 已移动消息 ${tid} 到备份目录"
                
                # 释放信号量
                echo >&6
            } &
        fi
    done
    
    wait
}

5. 结果汇总模块

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

#!/usr/bin/env bash

# 结果汇总与统计
result_summary() {
    local backup_dir="/opt/app/backup/$(date +%F)/"
    local log_file="/opt/app/logs/cleanup_$(date +%F).log"
    
    # 统计处理结果
    local msg_count=$(find "${backup_dir}/msg/" -type f | wc -l)
    local context_count=$(find "${backup_dir}/dacontext/" -type f | wc -l)
    
    # 记录汇总信息
    {
        echo "=========================================="
        echo "清理操作完成时间: $(date '+%F %H:%M:%S')"
        echo "处理消息数量: ${msg_count}"
        echo "处理上下文数量: ${context_count}"
        echo "备份目录: ${backup_dir}"
        echo "=========================================="
    } | tee -a "${log_file}"
}

关键Bash技巧

1. 管道与FIFO的巧妙使用

技巧要点：使用命名管道实现进程间通信，避免磁盘IO瓶颈

bash
1
2
3
4
5
6

# 创建双向通信管道
mkfifo /tmp/process_fifo && exec 3<> /tmp/process_fifo

# 使用管道进行异步通信
echo "task_data" >&3
read response <&3

实际应用：

  • 生产者向管道写入任务
  • 消费者从管道读取任务
  • 通过信号量控制并发数量

2. awk高效文本处理

技巧要点：使用awk进行复杂的文本解析和过滤

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 高效的日志解析
awk -F"[][]" '
$0 ~ /\[cmd:remote,tid/ {
    tid=gettid($2)
    result=getresult($2)
    if(result=="Normal-Spam"){
        print tid
    }
}
' /var/log/rcptstat.log

性能优势：

  • 单次遍历完成多重解析
  • 内置正则表达式引擎
  • 内存高效处理

3. 进程控制与信号量

技巧要点：使用文件描述符实现信号量机制

bash
1
2
3
4
5
6
7
8

# 创建信号量控制
for i in {1..10}; do echo; done >&6

# 获取信号量
read -u6

# 释放信号量  
echo >&6

4. 条件判断与错误处理

技巧要点：完善的错误检查和处理机制

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 安全的文件操作
if [[ -f "${file}" ]]; then
    process_file "${file}"
else
    log_error "文件不存在: ${file}"
fi

# 日志函数
log_error() {
    echo "[$(date '+%F %H:%M')] ERROR: $1" | tee -a "${LOG_FILE}"
}

5. 路径处理与变量安全

技巧要点：安全的路径构建和变量处理

bash
1
2
3
4
5
6
7

# 安全的路径拼接
BASE_DIR="/opt/app"
TARGET_DIR="${BASE_DIR}/backup/$(date +%F)"
mkdir -p "${TARGET_DIR}"

# 变量引用安全
echo "处理文件: ${file:-未指定}"

完整集成示例

主控制脚本

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

#!/usr/bin/env bash

#!/usr/bin/env bash
# 消息队列清理工具 - 生产者消费者模式
# 针对反垃圾引擎标记的消息进行批量清理

# 配置参数
PROCESS_NUMS=10
WARNING_THRESHOLD=4000
CRITICAL_THRESHOLD=10000
HELL_THRESHOLD=16000

# 基础路径配置
BASE_DIR="/opt/app"
LOG_DIR="${BASE_DIR}/logs"
BACKUP_DIR="${BASE_DIR}/backup/$(date +%F)"
MSG_DIR="${BASE_DIR}/var/mda/msg"
CONTEXT_DIR="${BASE_DIR}/var/mda/dacontext"
RCPT_LOG_DIR="${BASE_DIR}/logs/rcptstat"

# 日志文件
CLEANUP_LOG="${LOG_DIR}/cleanup_$(date +%F).log"
MOVE_LOG="${LOG_DIR}/move_$(date +%F).log"

# 创建必要的目录
mkdir -p "${LOG_DIR}" "${BACKUP_DIR}"{msg,dacontext}

# 日志函数
log_error() {
    echo "[$(date '+%F %H:%M:%S')] ERROR: $1" | tee -a "${CLEANUP_LOG}"
}

log_info() {
    echo "[$(date '+%F %H:%M:%S')] INFO: $1" | tee -a "${CLEANUP_LOG}"
}

# 初始化管道和信号量
init_pipes() {
    # 进程控制管道
    local process_fifo="/tmp/$$process.fifo"
    mkfifo "${process_fifo}" && exec 6<>"${process_fifo}" && rm "${process_fifo}"
    for ((i=0; i<${PROCESS_NUMS}; i++)); do echo; done >&6
    
    # 任务传递管道
    local task_fifo="/tmp/$$task.fifo"
    mkfifo "${task_fifo}" && exec 7<>"${task_fifo}" && rm "${task_fifo}"
}

# 清理函数
cleanup_functions() {
    # 清理管道
    exec 6>&-
    exec 7>&-
    log_info "清理完成"
}

# 主函数
main() {
    log_info "开始消息队列清理检查"
    
    # 初始化管道
    init_pipes
    
    # 检查队列状态
    local queue_count=$(find "${MSG_DIR}" -type f | wc -l)
    log_info "当前队列文件数量: ${queue_count}"
    
    # 根据队列状态选择处理策略
    if [[ ${queue_count} -gt ${HELL_THRESHOLD} ]]; then
        log_error "队列数量 ${queue_count} 已超过阈值 ${HELL_THRESHOLD}，启动地狱模式"
        emergency_cleanup
    elif [[ ${queue_count} -gt ${CRITICAL_THRESHOLD} ]]; then
        log_error "队列数量 ${queue_count} 已超过阈值 ${CRITICAL_THRESHOLD}，启动紧急模式"
        critical_cleanup
    elif [[ ${queue_count} -gt ${WARNING_THRESHOLD} ]]; then
        log_error "队列数量 ${queue_count} 已超过阈值 ${WARNING_THRESHOLD}，启动快速模式"
        quick_cleanup
    else
        log_info "队列数量正常，无需处理"
    fi
    
    # 清理资源
    cleanup_functions
    result_summary
}

# 执行主函数
main "$@"

总结

本文详细介绍了一套基于生产者消费者模式的Bash高级编程方案，用于解决消息队列积压问题。通过多进程并发、管道通信、awk高效解析等技术的综合运用，实现了高效的队列清理。

核心优势

  1. 性能优异：多进程并发处理，效率提升数倍
  2. 架构清晰：生产者消费者分离，易于维护和扩展
  3. 稳定可靠：完善的错误处理和资源管理机制
  4. 配置灵活：支持不同阈值和并发策略

适用场景

  • 反垃圾引擎标记的消息批量清理
  • 恶意攻击导致的队列积压处理
  • 系统故障后的队列恢复
  • 定期的队列维护和优化

扩展建议

  1. 监控集成：与监控系统集成，实现自动化告警和处理
  2. 策略优化：根据实际业务需求，调整并发数量和阈值
  3. 文档完善：建立详细的操作手册和应急预案
  4. 测试验证：定期进行压力测试，确保方案有效性

这套方案提供了一个可复用的架构模式，可根据实际消息处理场景调整并发数与阈值后复用。