发送批量通知时EC2 CPU占用过高的技术求助

问题背景 #

我在Laravel项目中集成AWS SQS实现异步通知：会员发布专属帖子时，需给5000+订阅者发送Twilio短信、邮件、Firebase推送，合计15000+条通知。但服务器处理时EC2 CPU瞬间飙升至99%，触发Nginx 500错误。

原代码模块 #

帖子创建队列分发 #

$membershipSubscriberIdsChunk = $membershipSubscriber->pluck("id")->chunk(30)->toArray();

foreach ($membershipSubscriberIdsChunk as $key => $chunk) {
    dispatch(new SendPostMailNotificationJob($chunk, $mailPost, $membershipTitle, 0))->onQueue("post-email");
    dispatch(new SendPostPushNotificationJob($chunk, $mailPost->user->name))->onQueue("post-firebase");
    dispatch(new SendPostSMSNotificationJob($chunk, $mailPost->user->name))->onQueue("post-sms");
}

Twilio短信发送 #

$recipients = $fans->filter(function ($item) {
    if ($item->receive_sms && isset($item->phone) && $item->phone) {
        $countrycode = $item->country_code ?? '+1';
        $phone = preg_replace("/[^0-9]/", "", $item->phone);
        $phone = substr($phone, -10);
        if ($phone) {
            $item->phoneNumber = $countrycode . $phone;
            return $item;
        }
    }
})->pluck("phoneNumber")->toArray();

$message = "$this->creatorName has added a new post.";
$messageOptions = [
    'from' => config('services.twilio.from_number'),
    'body' => $message,
];
Log::channel("post_sms")->info("Twilio SMS Notification Process Start at " . Carbon::now()->format("Y-m-d H:i:s"));
$twilio = new Client(config('services.twilio.sid'), config('services.twilio.auth_token'));
foreach ($recipients as $key => $recipient) {
    try {
        Log::channel("post_sms")->info("$key SMS Start");
        $twilioResponse = $twilio->messages->create($recipient, $messageOptions);
        Log::channel("post_sms")->info("$key [$recipient] SMS Done.");
    } catch (Exception $e) {
        Log::channel("post_sms")->error("$recipient has an error." . $e->getMessage());
    }
}
Log::channel("post_sms")->info("Twilio SMS Notification Process Completed at " . Carbon::now()->format("Y-m-d H:i:s") . ".");

return response()->json([
    "message" => "Sent Request to Twilio."
]);

Firebase推送通知 #

$subscribers = MembershipSubscriber::with(["fan"])->whereIn("id", $this->membershipSubscriber)->get();
Log::channel('post_firebase')->info("Firebase Notification Process Start at " . Carbon::now()->format("Y-m-d H:i:s"));
$message = $this->creatorName . " has added a new post.";
$title = "New Post Added";

if ($subscribers->count()) {
    foreach ($subscribers as $key => $subscriber) {
        if (isset($subscriber->fan) && isset($subscriber->fan->apn_token) && $subscriber->fan->apn_token != "") {
            $optionBuilder = new OptionsBuilder();
            $optionBuilder->setTimeToLive(60 * 20);

            $notificationBuilder = new PayloadNotificationBuilder($title);
            $notificationBuilder->setBody($message)->setSound('default');

            $dataBuilder = new PayloadDataBuilder();
            $dataBuilder->addData(['subscriber_id' => $subscriber->id, 'type' => 'post']);

            $option = $optionBuilder->build();
            $notification = $notificationBuilder->build();
            $data = $dataBuilder->build();

            $token = $subscriber->fan->apn_token;
            $downstreamResponse = FCM::sendTo($token, $option, $notification, $data);
            $responseData = [
                "numberSuccess" => $downstreamResponse->numberSuccess(),
                "numberFailure" => $downstreamResponse->numberFailure(),
                "numberModification" => $downstreamResponse->numberModification(),
            ];
            Log::channel('post_firebase')->info("response for token.", $responseData);
        }
    }
    Log::channel('post_firebase')->info("Firebase Notification Process Completed at " . Carbon::now()->format("Y-m-d H:i:s") . "");
    return response()->json([
        "message" => "Push Notifications sent successfully"
    ]);
}

邮件通知 #

$fans = MembershipSubscriber::with(["fan"])->whereIn("id", $this->membershipSubscriber)->get()->pluck("fan");
$fans = $fans->where("receive_email", 1);
if ($fans->count()) {
    Log::channel('post_mail')->info("Mail Notification Main Job Start at " . Carbon::now()->format("Y-m-d H:i:s"));
    Notification::send($fans, (new PostCreatedEmailNotification($this->membershipPost, $this->membershipTitle, $this->jobDelay)));
    Log::channel('post_mail')->info("Mail Notification Main Job Completed at " . Carbon::now()->format("Y-m-d H:i:s") . ".");
} else {
    Log::channel('post_mail')->error("Mail Notification Fan User Not Found.");
}

核心问题分析 #

1. 单Job内同步循环调用第三方API：每条通知都发起单独HTTP请求，阻塞进程并占用大量CPU/网络资源。
2. Chunk粒度不合理：30个订阅者分块导致生成167个Job，瞬间占用所有Worker进程，触发CPU过载。
3. 未利用批量接口：Twilio/FCM均支持批量发送，却用单条循环，效率极低。
4. 邮件通知未异步：Notification::send在Job内同步发送，未拆分到独立队列任务。

优化方案 #

1. 改用第三方批量发送接口 #

Twilio短信批量优化

// 替换原循环代码
try {
    Log::channel("post_sms")->info("批量SMS发送开始");
    // Twilio批量限制单请求最多100条，分块处理
    foreach (array_chunk($recipients, 100) as $recipientChunk) {
        $twilioResponse = $twilio->messages->create(
            $recipientChunk, // 直接传收件人数组
            $messageOptions
        );
        Log::channel("post_sms")->info("批次SMS完成，数量：".count($recipientChunk));
    }
} catch (Exception $e) {
    Log::channel("post_sms")->error("批量SMS失败：".$e->getMessage());
}

Firebase推送批量优化

// 先收集有效令牌
$tokens = [];
foreach ($subscribers as $subscriber) {
    if (isset($subscriber->fan) && !empty($subscriber->fan->apn_token)) {
        $tokens[] = $subscriber->fan->apn_token;
    }
}

if (!empty($tokens)) {
    Log::channel('post_firebase')->info("批量推送开始，令牌数：".count($tokens));
    // FCM单请求最多500个令牌，分块处理
    foreach (array_chunk($tokens, 500) as $tokenChunk) {
        $downstreamResponse = FCM::sendTo($tokenChunk, $option, $notification, $data);
        $responseData = [
            "numberSuccess" => $downstreamResponse->numberSuccess(),
            "numberFailure" => $downstreamResponse->numberFailure(),
            "numberModification" => $downstreamResponse->numberModification(),
        ];
        Log::channel('post_firebase')->info("批次推送响应", $responseData);
    }
}

2. 调整Job分块与队列并发 #

• 增大分块大小：将30个订阅者分块改为100-200个，减少Job总数，避免Worker瞬间过载。
• 限制队列并发数：在queue.php中配置SQS队列的最大并发：

'sqs' => [
    // ... 原有配置
    'max_concurrent_jobs' => 5, // 每个队列最多同时处理5个Job
],

或者启动Worker时指定并发：

php artisan queue:work sqs --queue=post-email,post-firebase,post-sms --tries=3 --timeout=60 --concurrency=5

3. 邮件通知异步化 #

修改PostCreatedEmailNotification类，实现ShouldQueue接口：

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Notifications\Notification;

class PostCreatedEmailNotification extends Notification implements ShouldQueue
{
    use Queueable;

    // ... 原有代码
}

此时Notification::send会自动将每个邮件任务拆分到队列，避免同步阻塞。

4. 减少数据库重复查询 #

分发Job时直接传递所需字段，避免每个Job重复查询数据库：

// 预加载并筛选所需数据
$membershipSubscribers = $membershipSubscriber->with(['fan' => function($query) {
    $query->select('id', 'email', 'receive_email', 'phone', 'country_code', 'apn_token', 'receive_sms');
}])->get();

$chunks = $membershipSubscribers->chunk(100);
foreach ($chunks as $chunk) {
    // 拆分短信、推送、邮件数据
    $smsData = $chunk->filter(function($item) {
        return $item->fan->receive_sms && !empty($item->fan->phone);
    })->map(function($item) {
        $countrycode = $item->fan->country_code ?? '+1';
        $phone = preg_replace("/[^0-9]/", "", $item->fan->phone);
        $phone = substr($phone, -10);
        return $countrycode . $phone;
    })->filter()->toArray();

    $pushData = $chunk->filter(function($item) {
        return !empty($item->fan->apn_token);
    })->pluck('fan.apn_token')->toArray();

    $emailData = $chunk->filter(function($item) {
        return $item->fan->receive_email;
    })->pluck('fan')->toArray();

    // 仅当有数据时分发Job
    if (!empty($smsData)) dispatch(new SendPostSMSNotificationJob($smsData, $mailPost->user->name))->onQueue("post-sms");
    if (!empty($pushData)) dispatch(new SendPostPushNotificationJob($pushData, $mailPost->user->name))->onQueue("post-firebase");
    if (!empty($emailData)) dispatch(new SendPostMailNotificationJob($emailData, $mailPost, $membershipTitle, 0))->onQueue("post-email");
}

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