czg_team/p_ysk
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

初步MQ

Browse Source
This commit is contained in:
ASUS
2025-10-14 15:10:13 +08:00
parent 8aed3e5aac
commit d2a46c4738
15 changed files with 512 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
app/command/ApplySmsTemp.php Normal file
View File

@@ -0,0 +1,82 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 申请新短信模版
class ApplySmsTemp extends Command
{
protected static $defaultName = 'applysmstemp';
protected static $defaultDescription = 'applysmstemp';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = RabbitMqConfig::$host;
$port = RabbitMqConfig::$port;
$user = RabbitMqConfig::$user;
$password = RabbitMqConfig::$password;
$queue = RabbitMqConfig::$queue_t . '-apply.sms.temp';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('收到消息--->' . $data . '--->' . $date_time . "\n");
// 发给队列
Redis::send('apply.sms.temp', $data);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

82
app/command/BirthdayGiftSms.php Normal file
View File

@@ -0,0 +1,82 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 生日有礼
class BirthdayGiftSms extends Command
{
protected static $defaultName = 'birthdaygiftsms';
protected static $defaultDescription = 'birthdaygiftsms';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = RabbitMqConfig::$host;
$port = RabbitMqConfig::$port;
$user = RabbitMqConfig::$user;
$password = RabbitMqConfig::$password;
$queue = RabbitMqConfig::$queue_t . '-birthday.gift.sms';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('收到消息--->' . $data . '--->' . $date_time . "\n");
// 发给队列
Redis::send('birthday.gift.sms', $data);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

39
app/command/QuerySmsStatus.php Normal file
View File

@@ -0,0 +1,39 @@
<?php
namespace app\command;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use ZipArchive;
// 查询短信发送状态
class QuerySmsStatus extends Command
{
protected static $defaultName = 'querysmsstatus';
protected static $defaultDescription = 'querysmsstatus';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$list = Db::table('sms_user_push_event')->where(['push_type' => 2])->select()->toArray();
return self::SUCCESS;
}
}

82
app/command/SendMarkSms.php Normal file
View File

@@ -0,0 +1,82 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 发送营销短信
class SendMarkSms extends Command
{
protected static $defaultName = 'sendmarksms';
protected static $defaultDescription = 'sendmarksms';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = RabbitMqConfig::$host;
$port = RabbitMqConfig::$port;
$user = RabbitMqConfig::$user;
$password = RabbitMqConfig::$password;
$queue = RabbitMqConfig::$queue_t . '-send.mark.sms';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('收到消息--->' . $data . '--->' . $date_time . "\n");
// 发给队列
Redis::send('send.mark.sms', $data);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

82
app/command/SendWechatTemp.php Normal file
View File

@@ -0,0 +1,82 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 发送微信模版消息
class SendWechatTemp extends Command
{
protected static $defaultName = 'sendwechattemp';
protected static $defaultDescription = 'sendwechattemp';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = RabbitMqConfig::$host;
$port = RabbitMqConfig::$port;
$user = RabbitMqConfig::$user;
$password = RabbitMqConfig::$password;
$queue = RabbitMqConfig::$queue_t . '-send.wechat.temp';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('收到消息--->' . $data . '--->' . $date_time . "\n");
// 发给队列
Redis::send('send.wechat.temp', $data);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

21
app/common/controller/ApiController.php Normal file
View File

@@ -0,0 +1,21 @@
<?php
namespace app\common\controller;
use app\exception\MyBusinessException;
use support\exception\BusinessException;
class ApiController
{
protected static $token = 'kiy7s194DXKSpP8CJtRpnYn5r5HrHQTf';
public function __construct()
{
$request = request();
$get_token = $request->header('token');
// 基类
if(empty($get_token)) {
throw new MyBusinessException('参数错误', 500);
}
if($get_token != self::$token) {
throw new MyBusinessException('参数错误', 500);
}
}
}

16
app/controller/IndexController.php → app/controller/YskController.php
View File

@@ -2,10 +2,15 @@
namespace app\controller;
use app\common\controller\ApiController;
use app\exception\MyBusinessException;
use app\model\AlibabaSms;
use support\Request;
use support\think\Db;
use support\think\Model;
class IndexController
use Webman\RedisQueue\Client;
class YskController extends ApiController
{
public function index(Request $request)
{
@@ -22,17 +27,8 @@ class IndexController
// $data = \app\model\MessagePushTask::send_push_msg();
// $data = \app\model\MessagePushTask::send_sms_msg();
return json(123);
}
public function view(Request $request)
{
return view('index/view', ['name' => 'webman']);
}
public function json(Request $request)
{
return json(['code' => 0, 'msg' => 'ok']);
}
}

16
app/exception/MyBusinessException.php Normal file
View File

@@ -0,0 +1,16 @@
<?php
namespace app\exception;
use support\exception\BusinessException;
use Webman\Http\Request;
use Webman\Http\Response;
class MyBusinessException extends BusinessException
{
public function render(Request $request): ?Response
{
// json请求返回json数据
return json(['code' => $this->getCode() ?: 500, 'message' => $this->getMessage()]);
}
}

12
app/model/AlibabaSms.php
View File

@@ -2,6 +2,7 @@
namespace app\model;
use support\think\Db;
use think\Exception;
use support\Log;
use think\Model;
@@ -32,6 +33,7 @@ class AlibabaSms extends Model
}
/**
* 发送短信
* @param array $args
* @return void
*/
@@ -43,6 +45,7 @@ class AlibabaSms extends Model
try {
// 复制代码运行请自行打印 API 的返回值
$res = $client->sendSmsWithOptions($sendSmsRequest, $runtime);
Log::info('短信接口返回---》' . json_encode($res->body->toArray()));
if($res->body->code == 'OK') {
return true;
}else {
@@ -60,6 +63,15 @@ class AlibabaSms extends Model
}
}
/**
* 查询短信发送结果
*/
public function QuerySendDetails($data)
{
}
}

22
app/queue/redis/ApplySmsTemp.php Normal file
View File

@@ -0,0 +1,22 @@
<?php
namespace app\queue\redis;
use app\model\AlibabaSms;
use support\Log;
use support\think\Db;
use Webman\RedisQueue\Consumer;
// 申请新短信模版
class ApplySmsTemp implements Consumer
{
public $queue = 'apply.sms.temp';
public function consume($data)
{
Log::info('消息队列【申请新短信模版】接收到消息' . $data);
}
}

21
app/queue/redis/BirthdayGiftSms.php Normal file
View File

@@ -0,0 +1,21 @@
<?php
namespace app\queue\redis;
use app\model\AlibabaSms;
use support\Log;
use support\think\Db;
use Webman\RedisQueue\Consumer;
// 生日有礼
class BirthdayGiftSms implements Consumer
{
public $queue = 'birthday.gift.sms';
public function consume($data)
{
Log::info('消息队列【生日有礼】接收到消息' . $data);
}
}

21
app/queue/redis/SendMarkSms.php Normal file
View File

@@ -0,0 +1,21 @@
<?php
namespace app\queue\redis;
use app\model\AlibabaSms;
use support\Log;
use support\think\Db;
use Webman\RedisQueue\Consumer;
// 发送营销短信
class SendMarkSms implements Consumer
{
public $queue = 'send.mark.sms';
public function consume($data)
{
Log::info('消息队列【发送营销短信】接收到消息' . $data);
}
}

21
app/queue/redis/SendWechatTemp.php Normal file
View File

@@ -0,0 +1,21 @@
<?php
namespace app\queue\redis;
use app\model\AlibabaSms;
use support\Log;
use support\think\Db;
use Webman\RedisQueue\Consumer;
// 发送短信模版
class SendWechatTemp implements Consumer
{
public $queue = 'send.wechat.temp';
public function consume($data)
{
Log::info('消息队列【发送微信模版消息】接收到消息' . $data);
}
}

4
config/plugin/webman/redis-queue/redis.php
View File

@@ -1,9 +1,9 @@
<?php
return [
'default' => [
'host' => 'redis://127.0.0.1:6379',
'host' => 'redis://192.168.1.42:6379',
'options' => [
'auth' => null, // 密码，字符串类型，可选参数
'auth' => 'Chaozg123.', // 密码，字符串类型，可选参数
'db' => 0, // 数据库
'prefix' => '', // key 前缀
'max_attempts' => 5, // 消费失败后，重试次数

6
config/redis.php
View File

@@ -14,14 +14,14 @@
return [
'default' => [
'password' => '222222',
'password' => 'Chaozg123.',
'host' => '192.168.1.42',
'port' => 6379,
'database' => 0,
// Connection pool, supports only Swoole or Swow drivers.
'pool' => [
'max_connections' => 5,
'min_connections' => 1,
'max_connections' => 100,
'min_connections' => 10,
'wait_timeout' => 3,
'idle_timeout' => 60,
'heartbeat_interval' => 50,