本文目录导读：
我来详细介绍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);
}
}
这个短信队列与重试系统提供了完整的解决方案,包括：
- 支持多种短信服务商
- 异步处理，提高系统响应速度
- 可靠的重试机制
- 失败的记录和告警
- 限流和防护机制
- 监控和统计功能
使用这个系统可以有效提高短信发送的可靠性和系统的稳定性。
