composer require php-amqplib/php-amqplib 3.1.2
<?php
namespace app\ospay\model;
use app\common\library\helper;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPSocketConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPAbstractCollection;
use PhpAmqpLib\Wire\AMQPTable;
use think\facade\Db;
use think\Model;
use app\ospay\serve\LvServe;
/**
* 支付系统
* Class OsPay
* @package app\ospay\model
*/
class OsQueen extends Model
{
public static $QUEEN_KEY = 'OS_PAY_NOTIFY_LIST';
public static $CON;
public static function getQueenCon()
{
// 创建链接
$connection = new AMQPSocketConnection(
config('rabbitmq.host'), config('rabbitmq.port'),
config('rabbitmq.login'), config('rabbitmq.password'),
config('rabbitmq.vhost'), false,
'AMQPLAIN', null, 'en_US',
3, false, 3, 60
);
return $connection;
}
public static function getChan(AMQPSocketConnection $connection)
{
$arg = new AMQPTable();
$arg->set('x-max-length', 100*10000); //缓冲队列 最长100万
$arg->set('x-max-length-bytes', 512 * 1000 * 1000); //单位字节（byte） 512M
$channel = $connection->channel();
$channel->queue_declare(OsQueen::$QUEEN_KEY, false, true, false, false, false, $arg);
return $channel;
}
/**
* 添加到队列
*
*/
public static function sendQueen(AMQPChannel $channel,$data)
{
$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($msg, '', self::$QUEEN_KEY);
}
/**
* 消费者
* @throws \ErrorException
*/
public static function consume(string $tag)
{
echo $tag;
// 创建链接
$con=OsQueen::getQueenCon();
$chan=OsQueen::getChan($con);
$func = function (AMQPMessage $msg) {
OsQueen::notifyOrder($msg->getBody());
$msg->ack();
};
$chan->basic_qos(null, 1, null);
$chan->basic_consume(self::$QUEEN_KEY, $tag, false, false, false, false, $func);
while($chan->is_open()) {
$chan->wait();
}
}
public static function notifyOrder($queen_id)
{
$order = OsQueen::find($queen_id);
$config=OsShopAttach::where('shop_id',$order['shop_id'])->cache(600)->find();
if(!empty($order)&&$order['notify_status']==0){
$data = [
'code' => $order['code'],
'shop_id' => $order['shop_id'],
'out_trade_no' => $order['out_trade_no'],
'order_no' => $order['order_no'],
'total_fee' => $order['total_fee'],
];
$data['sign'] = LvServe::sign($data,$config['md5_key']);
$data['attach'] = $order['attach'];
$data['msg'] = $order['msg'];
$w[] = [
'id', '=', $order['id']
];
try {
$resp = LvServe::curl($order['notify_url'], $data, 6);
$up = [];
$up['notify_status'] = strtoupper($resp) == 'SUCCESS' ? 1 : 0;
$up['update_at'] = time();
$up['notify_num']=Db::raw('notify_num+1');
OsQueen::where($w)->update($up);
} catch (\Exception $e) {
log_write(helper::jsonEncode($e->getMessage()), 'error');
$up['notify_status'] = 0;
$up['update_at'] = time();
OsQueen::where($w)->update($up);
return;
}
}
}
}