本文目录导读：
我来详细介绍ThinkPHP项目中短信队列与重试机制的实现方案。
整体架构设计
基础配置
// config/sms.php
return [
'default' => 'aliyun',
'channels' => [
'aliyun' => [
'access_key' => env('SMS_ALIYUN_ACCESS_KEY'),
'access_secret' => env('SMS_ALIYUN_ACCESS_SECRET'),
'sign_name' => env('SMS_ALIYUN_SIGN_NAME'),
'template_code' => env('SMS_ALIYUN_TEMPLATE_CODE'),
],
'tencent' => [
'secret_id' => env('SMS_TENCENT_SECRET_ID'),
'secret_key' => env('SMS_TENCENT_SECRET_KEY'),
'app_id' => env('SMS_TENCENT_APP_ID'),
'sign_name' => env('SMS_TENCENT_SIGN_NAME'),
],
],
'queue' => [
'enabled' => true,
'connection' => 'redis',
'queue_name' => 'sms_queue',
],
'retry' => [
'max_attempts' => 3,
'retry_seconds' => [1, 5, 15], // 递增重试间隔
'max_failures' => 5, // 最大失败记录
],
];
短信服务类实现
短信服务基类
<?php
namespace app\common\service\sms;
use think\facade\Log;
use think\Exception;
abstract class SmsService
{
protected $config;
protected $queue;
public function __construct()
{
$this->config = config('sms');
$this->queue = app('sms.queue');
}
/**
* 抽象方法：发送短信
*/
abstract protected function sendRequest($phone, $templateCode, $params);
/**
* 统一发送接口
*/
public function send($phone, $templateCode, $params = [])
{
// 短信验证码场景特殊处理
if ($templateCode == 'verification_code') {
return $this->sendWithCode($phone, $params);
}
// 是否启用队列
if ($this->config['queue']['enabled']) {
return $this->pushToQueue($phone, $templateCode, $params);
}
// 直接发送
return $this->sendWithRetry($phone, $templateCode, $params);
}
/**
* 将短信加入队列
*/
protected function pushToQueue($phone, $templateCode, $params)
{
$jobData = [
'phone' => $phone,
'template_code' => $templateCode,
'params' => $params,
'created_at' => time(),
'attempts' => 0,
];
$queueConfig = $this->config['queue'];
$queue = Queue::connection($queueConfig['connection']);
// 延迟队列支持
$delay = $params['delay'] ?? 0;
unset($params['delay']);
return $queue->later(
$delay,
'app\common\job\SmsJob',
$jobData,
$queueConfig['queue_name']
);
}
/**
* 带重试机制的发送
*/
protected function sendWithRetry($phone, $templateCode, $params)
{
$maxAttempts = $this->config['retry']['max_attempts'];
$retrySeconds = $this->config['retry']['retry_seconds'];
for ($attempt = 1; $attempt <= $maxAttempts; $attempt++) {
try {
$result = $this->sendRequest($phone, $templateCode, $params);
if ($result['success']) {
$this->logSuccess($phone, $templateCode, $attempt);
return $result;
}
// 如果服务商明确表示失败，不重试
if (isset($result['retryable']) && !$result['retryable']) {
break;
}
} catch (\Exception $e) {
Log::error("短信发送异常: " . $e->getMessage(), [
'phone' => $phone,
'template' => $templateCode,
'attempt' => $attempt
]);
}
// 重试间隔
if ($attempt < $maxAttempts) {
$delay = $retrySeconds[$attempt - 1] ?? $retrySeconds[count($retrySeconds) - 1];
sleep($delay);
}
}
// 发送失败，记录并通知
$this->handleFailure($phone, $templateCode, $params);
return ['success' => false, 'message' => '短信发送失败'];
}
/**
* 发送验证码
*/
protected function sendWithCode($phone, $params)
{
$code = $params['code'] ?? $this->generateCode();
$expireMinutes = $params['expire_minutes'] ?? 10;
// 存储验证码
cache("sms_code:{$phone}", [
'code' => $code,
'expire_time' => time() + $expireMinutes * 60
], $expireMinutes * 60);
return $this->send($phone, 'verification_code', [
'code' => $code
]);
}
/**
* 生成验证码
*/
protected function generateCode($length = 6)
{
$characters = '0123456789';
$code = '';
for ($i = 0; $i < $length; $i++) {
$code .= $characters[rand(0, strlen($characters) - 1)];
}
return $code;
}
/**
* 记录成功日志
*/
protected function logSuccess($phone, $template, $attempt)
{
Log::info("短信发送成功", [
'phone' => $phone,
'template' => $template,
'attempt' => $attempt,
'time' => date('Y-m-d H:i:s')
]);
}
/**
* 处理发送失败
*/
protected function handleFailure($phone, $template, $params)
{
// 记录到数据库
$smsLog = new SmsLog();
$smsLog->phone = $phone;
$smsLog->template_code = $template;
$smsLog->params = json_encode($params);
$smsLog->status = 'failed';
$smsLog->error_message = '发送失败';
$smsLog->save();
// 发送告警通知
$this->sendAlert('sms_failed', [
'phone' => $phone,
'template' => $template
]);
}
}
阿里云短信实现
<?php
namespace app\common\service\sms;
use AlibabaCloud\Client\AlibabaCloud;
class AliyunSmsService extends SmsService
{
protected function sendRequest($phone, $templateCode, $params)
{
$config = $this->config['channels']['aliyun'];
try {
AlibabaCloud::accessKeyClient(
$config['access_key'],
$config['access_secret']
)
->regionId('cn-hangzhou')
->asDefaultClient();
$result = AlibabaCloud::rpc()
->product('Dysmsapi')
->version('2017-05-25')
->action('SendSms')
->method('POST')
->options([
'query' => [
'PhoneNumbers' => $phone,
'SignName' => $config['sign_name'],
'TemplateCode' => $templateCode,
'TemplateParam' => json_encode($params),
],
])
->request();
$response = $result->toArray();
return [
'success' => $response['Code'] == 'OK',
'message' => $response['Message'] ?? '',
'request_id' => $response['RequestId'] ?? ''
];
} catch (\Exception $e) {
return [
'success' => false,
'message' => $e->getMessage(),
'retryable' => true
];
}
}
}
任务队列实现
短信任务类
<?php
namespace app\common\job;
use think\queue\Job;
use app\common\service\sms\SmsFactory;
use think\facade\Cache;
use think\facade\Log;
class SmsJob
{
/**
* 执行任务
*/
public function fire(Job $job, $data)
{
// 检查任务是否已被取消
if ($job->attempts() > 3) {
$job->delete();
return;
}
$retryConfig = config('sms.retry');
$maxAttempts = $retryConfig['max_attempts'];
if ($job->attempts() >= $maxAttempts) {
// 超过最大重试次数，记录失败
$this->markAsFailed($data, "超过最大重试次数");
$job->delete();
return;
}
try {
// 获取短信服务
$smsService = SmsFactory::create();
// 发送短信
$result = $smsService->sendDirect(
$data['phone'],
$data['template_code'],
$data['params']
);
if ($result['success']) {
// 发送成功，删除任务
$job->delete();
// 记录成功日志
Log::info("队列短信发送成功", [
'phone' => $data['phone'],
'template' => $data['template_code']
]);
} else {
// 发送失败，重新放入队列
$this->retryJob($job, $data, $result['message']);
}
} catch (\Exception $e) {
// 异常处理
Log::error("队列短信发送异常: " . $e->getMessage(), [
'job_id' => $job->getJobId(),
'data' => $data
]);
$this->retryJob($job, $data, $e->getMessage());
}
}
/**
* 重试任务
*/
protected function retryJob(Job $job, $data, $errorMessage = '')
{
$attempts = $job->attempts();
$delay = $this->getRetryDelay($attempts);
// 释放任务并设置延迟
$job->release($delay);
// 记录重试日志
Log::warning("短信任务重试", [
'attempts' => $attempts,
'delay' => $delay,
'error' => $errorMessage,
'phone' => $data['phone']
]);
}
/**
* 获取重试延迟时间
*/
protected function getRetryDelay($attempts)
{
$retrySeconds = config('sms.retry.retry_seconds');
$index = min($attempts - 1, count($retrySeconds) - 1);
return $retrySeconds[$index] ?? 10;
}
/**
* 标记为失败
*/
protected function markAsFailed($data, $reason)
{
// 记录失败日志
Log::error("短信任务失败", [
'phone' => $data['phone'],
'template' => $data['template_code'],
'reason' => $reason
]);
// 保存失败记录到数据库
SmsLog::create([
'phone' => $data['phone'],
'template_code' => $data['template_code'],
'params' => json_encode($data['params']),
'status' => 'failed',
'error_message' => $reason
]);
// 发送告警通知
$this->sendAlert('sms_queue_failed', [
'phone' => $data['phone'],
'reason' => $reason
]);
}
/**
* 任务失败回调
*/
public function failed($data, $e)
{
// 最后的重试也失败
Log::error("短信任务最终失败: " . $e->getMessage(), [
'data' => $data
]);
}
}
队列配置文件
// config/queue.php
return [
'default' => env('queue.driver', 'redis'),
'connections' => [
'redis' => [
'type' => 'redis',
'host' => env('REDIS_HOST', '127.0.0.1'),
'port' => env('REDIS_PORT', 6379),
'password' => env('REDIS_PASSWORD', ''),
'select' => env('REDIS_DB', 0),
'timeout' => 0,
'persistent' => false,
],
'database' => [
'type' => 'database',
'queue' => 'default',
'table' => 'jobs',
],
],
'failed' => [
'type' => 'database',
'table' => 'failed_jobs',
],
];
短信工厂与辅助类
短信工厂
<?php
namespace app\common\service\sms;
class SmsFactory
{
/**
* 创建短信服务
*/
public static function create($channel = null)
{
$channel = $channel ?: config('sms.default');
switch ($channel) {
case 'aliyun':
return new AliyunSmsService();
case 'tencent':
return new TencentSmsService();
default:
throw new \Exception("不支持的短信服务：" . $channel);
}
}
}
短信发送控制器
<?php
namespace app\api\controller;
use think\Controller;
use app\common\service\sms\SmsFactory;
use think\facade\Validate;
use think\facade\Cache;
class SmsController extends Controller
{
protected $smsService;
public function __construct()
{
$this->smsService = SmsFactory::create();
}
/**
* 发送验证码
*/
public function sendSms()
{
$validate = Validate::rule([
'phone' => 'require|mobile',
'type' => 'require|in:register,login,forget,change_phone',
]);
if (!$validate->check(input())) {
return json(['code' => 1, 'message' => $validate->getError()]);
}
$phone = input('phone');
$type = input('type');
// 防止重复发送
$key = "sms_send_limit:{$type}:{$phone}";
if (!Cache::has($key)) {
// 设置发送间隔限制 60秒
Cache::set($key, 1, 60);
} else {
return json(['code' => 1, 'message' => '发送太频繁，请稍后再试']);
}
// 生成验证码
$code = rand(100000, 999999);
// 发送短信
$result = $this->smsService->send($phone, 'verification_code', [
'code' => $code,
'delay' => 0,
]);
return json([
'code' => $result['success'] ? 0 : 1,
'message' => $result['success'] ? '验证码已发送' : $result['message']
]);
}
}
数据库表设计
短信日志表
CREATE TABLE `sms_logs` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`phone` varchar(20) NOT NULL COMMENT '手机号',
`template_code` varchar(50) NOT NULL COMMENT '模板编码',
`params` text COMMENT '模板参数',
`status` enum('pending','sending','success','failed') DEFAULT 'pending' COMMENT '状态',
`attempts` int(11) DEFAULT 0 COMMENT '尝试次数',
`error_message` text COMMENT '错误信息',
`request_id` varchar(100) DEFAULT NULL COMMENT '请求ID',
`created_at` datetime DEFAULT CURRENT_TIMESTAMP,
`updated_at` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_phone` (`phone`),
KEY `idx_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
失败任务表
CREATE TABLE `failed_jobs` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`connection` varchar(50) NOT NULL,
`queue` varchar(50) NOT NULL,
`payload` text NOT NULL,
`exception` text NOT NULL,
`failed_at` datetime NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
命令行操作
启动队列消费
# 启动短信队列消费进程 php think queue:work --queue sms_queue --tries 3
查看队列状态
// 控制器: QueueController.php
public function status()
{
$queueSize = Cache::get('sms_queue_size');
$failedJobs = FailedJob::count();
return json([
'queue_size' => $queueSize,
'failed_jobs' => $failedJobs,
'processed_count' => SmsLog::where('status', 'success')->count()
]);
}
高级特性
短信发送限流
// 使用令牌桶算法限流
class SmsRateLimiter
{
protected $bucketSize = 100; // 桶容量
protected $refillRate = 10; // 每秒填充速率
public function isAllowed($phone)
{
$key = "sms_rate:{$phone}";
$tokens = Cache::get($key);
if (!$tokens) {
$tokens = $this->bucketSize;
}
// 补充令牌
$lastRefill = Cache::get($key . ':time', time());
$elapsedSeconds = time() - $lastRefill;
$tokens = min(
$this->bucketSize,
$tokens + $elapsedSeconds * $this->refillRate
);
if ($tokens >= 1) {
Cache::set($key, $tokens - 1);
Cache::set($key . ':time', time());
return true;
}
return false;
}
}
多通道故障转移
class SmsService extends SmsService
{
public function sendWithFailover($phone, $templateCode, $params)
{
$channels = ['aliyun', 'tencent']; // 通道优先级
foreach ($channels as $channel) {
try {
$service = SmsFactory::create($channel);
$result = $service->sendDirect($phone, $templateCode, $params);
if ($result['success']) {
return $result;
}
Log::warning("短信通道失败，尝试下一个", [
'channel' => $channel,
'result' => $result
]);
} catch (\Exception $e) {
Log::error("短信通道异常", [
'channel' => $channel,
'error' => $e->getMessage()
]);
}
}
throw new \Exception('所有短信通道均不可用');
}
}
监控与告警
队列监控
class SmsQueueMonitor
{
public function checkQueueHealth()
{
$redis = new \Redis();
$redis->connect('127.0.0.1', 6379);
// 获取队列长度
$queueSize = $redis->lLen('queues:sms_queue');
// 检查延迟
$delaySize = $redis->lLen('queues:sms_queue_delayed');
// 检查是否有消费者
$consumerStatus = $redis->get('sms_queue:consumer_status');
return [
'queue_size' => $queueSize,
'delay_size' => $delaySize,
'consumer_running' => $consumerStatus === 'running',
'health' => ($queueSize < 1000 && $consumerStatus === 'running')
];
}
public function sendAlert($type, $message)
{
// 通过短信，邮件或Webhook发送告警
// 这里以邮件为例
mail('admin@example.com', "短信系统告警: {$type}", $message);
}
}
这个短信队列与重试系统提供了完整的解决方案,包括：
- 支持多种短信服务商
- 异步处理，提高系统响应速度
- 可靠的重试机制
- 失败的记录和告警
- 限流和防护机制
- 监控和统计功能
使用这个系统可以有效提高短信发送的可靠性和系统的稳定性。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
