高流量PHP脚本卸载AWS SNS消息至其他脚本/服务器的方案咨询

针对你每分钟1000+请求的PHP脚本发送AWS SNS消息的性能问题，直接给你几个可行的方案，按优先级排序：

1. 优先选择Redis消息队列（最优解） #

用Redis的List结构做消息队列是最适合的方案，完全不会阻塞主请求，性能拉满，可靠性也有保障。

实现思路： #

• 主请求脚本：把SNS消息的结构化数据（比如TopicARN、消息内容）序列化为JSON，用Redis的LPUSH原子推送到队列，操作耗时微秒级，不影响请求响应速度。
• 独立消费者脚本：用CLI模式运行（配合Supervisor守护进程保证一直运行），用BRPOP阻塞式从队列弹出消息，调用AWS SNS SDK发送消息；处理失败的消息可以放到单独的失败队列，后续重试。

代码示例： #

主请求脚本片段：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 构造SNS消息数据
$snsMsg = json_encode([
    'topic_arn' => 'arn:aws:sns:us-east-1:123456789012:YourTopic',
    'message' => '请求相关的消息内容',
    'subject' => '可选主题'
]);

// 推送到队列，无需等待结果
$redis->lpush('sns_task_queue', $snsMsg);

消费者脚本：

require 'vendor/autoload.php';

use Aws\Sns\SnsClient;
use Aws\Exception\AwsException;

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 初始化AWS SNS客户端
$sns = new SnsClient([
    'version' => 'latest',
    'region' => 'us-east-1',
    'credentials' => [
        'key' => '你的AWS Access Key',
        'secret' => '你的AWS Secret Key',
    ]
]);

while (true) {
    // 阻塞等待队列消息，避免空轮询浪费资源
    $result = $redis->brpop('sns_task_queue', 5); // 5秒超时后继续循环
    if ($result) {
        $msgData = json_decode($result[1], true);
        try {
            $sns->publish([
                'TopicArn' => $msgData['topic_arn'],
                'Message' => $msgData['message'],
                'Subject' => $msgData['subject'] ?? ''
            ]);
        } catch (AwsException $e) {
            // 处理失败，将消息放入失败队列（可后续重试）
            $redis->lpush('sns_failed_queue', $result[1]);
            error_log('SNS发送失败: ' . $e->getAwsErrorMessage());
        }
    }
}

2. 原子文件移动方案（无Redis时备选） #

如果无法使用Redis，这种方案比直接追加文件更安全，避免读写阻塞和数据丢失。

实现思路： #

• 主请求脚本：先在临时目录生成唯一命名的文件，写入消息内容，再通过rename原子操作将文件移动到指定的处理目录（rename在大多数文件系统中是原子操作，不会出现半写文件）。
• 消费者脚本：扫描处理目录，将文件先移到"处理中"临时位置（避免多消费者重复处理），读取内容发送SNS，成功后归档文件，失败则移回处理目录重试。

代码示例： #

主请求脚本片段：

$tempDir = '/tmp/sns_temp';
$processDir = '/var/www/sns_process';

// 确保目录存在
!is_dir($tempDir) && mkdir($tempDir, 0755, true);
!is_dir($processDir) && mkdir($processDir, 0755, true);

// 生成唯一临时文件
$tempFile = $tempDir . '/sns_msg_' . uniqid('', true) . '.tmp';
file_put_contents($tempFile, json_encode($snsMsg) . "\n");

// 原子移动到处理目录
rename($tempFile, $processDir . '/' . basename($tempFile));

消费者脚本片段：

$processDir = '/var/www/sns_process';
$archiveDir = '/var/www/sns_archive';
!is_dir($archiveDir) && mkdir($archiveDir, 0755, true);

// 初始化SNS客户端...

while (true) {
    $files = scandir($processDir);
    foreach ($files as $file) {
        if ($file === '.' || $file === '..') continue;
        $srcPath = $processDir . '/' . $file;
        $processingPath = $processDir . '/processing_' . $file;

        // 原子移动到处理中路径，避免竞争
        if (!rename($srcPath, $processingPath)) {
            continue;
        }

        $msgData = json_decode(trim(file_get_contents($processingPath)), true);
        try {
            $sns->publish([...]);
            // 处理成功，归档文件
            rename($processingPath, $archiveDir . '/' . date('Ymd_H') . '_' . $file);
        } catch (Exception $e) {
            // 处理失败，移回原目录重试
            rename($processingPath, $srcPath);
            error_log('SNS发送失败（文件：' . $file . '）: ' . $e->getMessage());
        }
    }
    sleep(1); // 降低目录扫描频率
}

3. 追加文件方案（简单但需注意细节） #

如果一定要用单文件追加，依赖文件系统的原子追加特性（如ext4、XFS），确保每次写入完整一行，消费者通过记录读取偏移量来避免重复或漏读。

注意事项： #

• 主脚本用非阻塞排他锁（LOCK_EX | LOCK_NB），获取不到锁时重试1-2次，避免阻塞请求。
• 消费者脚本记录上次读取到的文件大小，每次读取新增的内容，按行解析消息。

代码示例（主脚本）： #

$logFile = '/var/www/sns_messages.log';
$msgLine = json_encode($snsMsg) . "\n";

$handle = fopen($logFile, 'a');
if ($handle) {
    // 尝试获取非阻塞排他锁
    if (flock($handle, LOCK_EX | LOCK_NB)) {
        fwrite($handle, $msgLine);
        flock($handle, LOCK_UN);
    } else {
        // 锁获取失败，可重试1次或记录到临时存储
        file_put_contents('/tmp/sns_fallback.log', $msgLine, FILE_APPEND);
    }
    fclose($handle);
}

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