PHP项目消息批量推送如何优化性能

wen PHP项目 2026-06-30 26

本文目录导读：

PHP项目消息批量推送如何优化性能

核心架构优化：异步处理
发送过程优化
并发优化
缓存与预处理
性能监控与限流
实战配置示例
性能对比
建议优先级

针对PHP项目的消息批量推送性能优化,核心思路是避免逐个同步推送，改为异步、批量、并发处理，以下是具体的优化策略，从架构到代码实现：

核心架构优化：异步处理

1 消息队列（推荐方案）

将推送任务写入队列,由独立消费者处理：

// 生产者：写入队列
public function pushBatch(array $users, $message)
{
    foreach ($users as $user) {
        Redis::lPush('push_queue', json_encode([
            'user_id' => $user['id'],
            'message' => $message,
            'channel' => $user['channel'] // email/sms/push
        ]));
    }
}
// 消费者：独立进程批量消费
while (true) {
    $tasks = Redis::rPop('push_queue', 100); // 批量取出
    $batch = [];
    foreach ($tasks as $task) {
        $batch[$task['channel']][] = $task;
    }
    // 按渠道批量发送
    foreach ($batch as $channel => $items) {
        sendBatchByChannel($channel, $items);
    }
}

2 使用Swoole/Workerman

// Swoole异步推送
$server = new Swoole\Server('0.0.0.0', 9501);
$server->on('Receive', function ($server, $fd, $reactor_id, $data) {
    $tasks = json_decode($data, true);
    // 使用协程批量发送
    $channels = [];
    foreach ($tasks as $task) {
        $channels[$task['channel']][] = $task;
    }
    foreach ($channels as $channel => $items) {
        go(function() use ($items, $channel) {
            sendBatchByChannel($channel, $items);
        });
    }
});

发送过程优化

1 HTTP请求合并

对于API推送（如极光推送、个推），利用批量接口：

// 错误方式：逐个推送
foreach ($users as $user) {
    Http::post($pushApi, ['tokens' => [$user['token']], 'msg' => $msg]);
}
// 优化方式：批量推送
$chunks = array_chunk($userTokens, 500); // 一批最多500个
foreach ($chunks as $tokens) {
    Http::post($pushApi, ['tokens' => $tokens, 'msg' => $msg]);
}

2 数据库批量操作

// 批量插入推送记录
$records = [];
foreach ($users as $user) {
    $records[] = [
        'user_id' => $user['id'],
        'message' => $message,
        'status' => 0,
        'created_at' => date('Y-m-d H:i:s')
    ];
}
// 批次插入（500-1000条一次）
DB::table('push_logs')->insert($records);

并发优化

1 Guzzle并发请求

use GuzzleHttp\Client;
use GuzzleHttp\Pool;
$client = new Client(['timeout' => 3]);
$requests = function ($users) {
    foreach ($users as $user) {
        yield function() use ($client, $user) {
            return $client->postAsync($pushUrl, [
                'json' => ['token' => $user['token'], 'msg' => $msg]
            ]);
        };
    }
};
$pool = new Pool($client, $requests($users), [
    'concurrency' => 50, // 并发数
    'fulfilled' => function ($response, $index) {
        // 处理成功
    },
    'rejected' => function ($reason, $index) {
        // 处理失败
    },
]);
$promise = $pool->promise();
$promise->wait();

2 多进程处理

// 使用pcntl_fork
$workers = 10;
$chunks = array_chunk($users, ceil(count($users) / $workers));
foreach ($chunks as $i => $chunk) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('could not fork');
    } else if ($pid) {
        // 父进程
        $pids[] = $pid;
    } else {
        // 子进程：处理当前chunk
        foreach ($chunk as $user) {
            sendPush($user, $message);
        }
        exit(0);
    }
}
// 等待所有子进程结束
foreach ($pids as $pid) {
    pcntl_waitpid($pid, $status);
}

缓存与预处理

1 设备Token缓存

// 使用Redis Hash缓存用户设备信息
public function getUserDevice($userId)
{
    $key = "user_device:{$userId}";
    $device = Redis::hGetAll($key);
    if (empty($device)) {
        $device = DB::table('user_devices')->where('user_id', $userId)->first();
        Redis::hMset($key, (array)$device);
        Redis::expire($key, 3600); // 缓存1小时
    }
    return $device;
}

2 消息模板预编译

// 预编译模板，避免重复解析
$compiled = [];
foreach ($templates as $type => $template) {
    $compiled[$type] = new \Twig\Template\Template($template);
}
// 使用时直接渲染
foreach ($users as $user) {
    $msg = $compiled[$user['type']]->render(['name' => $user['name']]);
}

性能监控与限流

1 速率控制

// 使用Redis限制每秒发送量
$rateLimiter = new \RedisRateLimiter\RateLimiter(Redis::connection());
$maxPerSecond = 100;
foreach ($users as $user) {
    $rateLimiter->consume('push_rate', $maxPerSecond, 1); // 每秒最多100次
    sendPush($user, $message);
}

2 失败重试机制

// 失败任务入重试队列
function sendWithRetry($user, $message, $retryCount = 3) {
    try {
        sendPush($user, $message);
    } catch (\Exception $e) {
        if ($retryCount > 0) {
            Redis::rPush('push_retry', json_encode([
                'user' => $user,
                'message' => $message,
                'retry_count' => $retryCount - 1,
                'last_error' => $e->getMessage()
            ]));
        } else {
            // 记录最终失败
            DB::table('push_failed')->insert([...]);
        }
    }
}

实战配置示例

1 优化后的推送流程

[Web Server] → [Redis Queue] → [Worker进程] → [批量发送] → [推送服务]
     ↓                                                            ↓
  返回成功                                                    [日志记录]

2 Nginx配置优化

# 增加worker连接数
worker_connections 10240;
# PHP-FPM调优
pm.max_children = 100
pm.start_servers = 20
pm.min_spare_servers = 10
pm.max_spare_servers = 30

性能对比

方案	1000用户耗时	资源消耗	适用场景
同步逐个推送	30-60秒	CPU高	不推荐
Guzzle并发50	3-5秒	内存中等	小规模
消息队列+Worker	2-3秒	可控	中等规模
Swoole协程	1-2秒	极低	大规模

建议优先级

必须做：异步队列、批量发送（合并请求）
推荐做：并发请求（Guzzle Pool）、数据库批量插入
优化做：缓存、预编译、多进程
进阶做：Swoole/Workerman、分布式消费

选择方案时,根据你的用户规模（100、1000、10000+）和实时性要求来决定，对于大多数业务场景，Redis队列 + 批量推送 是最佳性价比方案。