<?php
namespace think;
date_default_timezone_set('Asia/Shanghai');
// 加载基础文件
require_once __DIR__ . '/thinkphp/base.php';
// 支持事先使用静态方法设置Request对象和Config对象
// 执行应用并响应
//Container::get('app')->run()->send();
//require_once __DIR__ . '/../../../thinkphp/helper.php';
use think\cache\driver\Redis;
//use think\Controller;
use think\Db;
class Swoole
{
const errcode = array(
43004 => '需要接收者关注',
40037 => '无效模板',
40003 => '需要接收者关注',
43005 => '需要好友关系',
43019 => '需要将接收者从黑名单中移除',
44001 => '多媒体文件为空',
44002 => 'POST 的数据包为空',
44003 => '图文消息内容为空',
44004 => '文本消息内容为空',
45001 => '多媒体文件大小超过限制',
45002 => '消息内容超过限制',
45003 => '标题字段超过限制',
45004 => '描述字段超过限制',
45005 => '链接字段超过限制',
45006 => '图片链接字段超过限制',
45007 => '语音播放时间超过限制',
45008 => '图文消息超过限制',
45009 => '接口调用超过限制',
45010 => '创建菜单个数超过限制',
45011 => 'API 调用太频繁，请稍候再试',
);
private $serv;
private $redis;
private $conn = [
// 数据库类型
'type' => 'mysql',
// 服务器地址
'hostname' => '',
// 数据库名
'database' => '',
// 用户名
'username' => '',
// 密码
'password' => '',
// 端口
'hostport' => '3306',
// 连接dsn
'dsn' => '',
// 数据库连接参数
'params' => [],
// 数据库编码默认采用utf8
'charset' => 'utf8',
// 数据库表前缀
'prefix' => 'shd_',
// 数据库调试模式
'debug' => true,
// 数据集返回类型
'resultset_type' => 'array',
// 自动写入时间戳字段
'auto_timestamp' => false,
// 时间字段取出后的默认时间格式
'datetime_format' => 'Y-m-d H:i:s',
// 是否需要进行SQL性能分析
'sql_explain' => false,
// Builder类
'builder' => '',
// Query类
'query' => '\\think\\db\\Query',
// 是否需要断线重连
'break_reconnect' => false,
// 断线标识字符串
'break_match_str' => [],
];
//初始化配置，监听端口
public function __construct()
{
//redis
$this->redis = new Redis();
$this->serv = new \swoole_server("0.0.0.0", 9501);
$this->serv->set(array(
'worker_num' => 2, //一般设置为服务器CPU数的1-4倍
'daemonize' => 1, //以守护进程执行
'max_request' => 10000,
'dispatch_mode' => 2,
'task_worker_num' => 8, //task进程的数量
"task_ipc_mode " => 3, //使用消息队列通信，并设置为争抢模式
"log_file" => "taskqueueu.log" ,//日志
));
$this->serv->on('Receive', array($this, 'onReceive'));
// bind callback
$this->serv->on('Task', array($this, 'onTask'));
$this->serv->on('Finish', array($this, 'onFinish'));
$this->serv->start();
}
//接收客户端的请求并响应
public function onReceive(\swoole_server $serv, $fd, $from_id, $data)
{
echo "Get Message From Client {$fd}:{$data}\n";
$serv->send($fd, '发送任务已建立，正在发送，请稍后查看发送记录');
// send a task to task worker.
$serv->task($data);//投递任务
}
public function onTask($serv, $task_id, $from_id, $data)
{
echo "Task {$task_id} task\n";
$array = json_decode($data, true);
$success = 0;
$fail = 0;
$log = '';
$access_token = $array['access_token'];
$openid_list = $this->redis->sMembers($array['appid'].'users');//从redis取出要批量发送的openid
$fields = json_decode($array['data'],true);
$send_data = array();
$start = time();
//模板消息
foreach ($openid_list as $openid) {
$template = array(
'touser' => $openid,
'template_id' => $array['tem_id'],
'url' => $array['url'],
'topcolor' => "#000000",
'data' => $send_data,
);
$url = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=" . $access_token;
$res = $this->send_post($url, $template);
$res_arr = json_decode($res, true);
if ($res_arr['errcode'] == 0){
++ $success;
}else{
++ $fail;
$log = self::errcode[$res_arr['errcode']];
}
}
$result = array('success'=>$success,'fail'=>$fail,'tem_id'=>$array['tem_id'],'uid'=>$array['uid'],'data'=>$array['data'],'url'=>$array['url'],'log'=>$log,'start'=>$start);
return json_encode($result);
}
//任务执行完自动回调结束方法
public function onFinish($serv, $task_id, $data)
{
$array = json_decode($data,true);
$fields = json_decode($array['data'],true);
//获取当前模板
$list = Db::connect($this->conn)->name('wechat_template')->where('template_id',$array['tem_id'])->where('uid',$array['uid'])->find();
$new_field = $list['field'];
$insert['template_id'] = $array['tem_id'];
$insert['success'] = $array['success'];
$insert['fail'] = $array['fail'];
$insert['url'] = $array['url'];
$insert['log'] = $array['log'];
$insert['create_time'] = date('Y-m-d H:i:s',$array['start']);
$insert['finish_time'] = date('Y-m-d H:i:s');
Db::connect($this->conn)->name('wechat_template_log')->insert($insert);
echo "Task{$data} {$task_id} finish\n";
}
function send_post($url, $post_data) {
$postdata=json_encode($post_data,JSON_UNESCAPED_UNICODE);
$options = array(
'http' => array(
'method' => 'POST',
'header' => 'Content-type:application/x-www-form-urlencoded',
'content' => $postdata,
// 'protocol_version' => 1.1,
// 'header' => [
// 'Connection: close',
// ],
'timeout' => 2 // 超时时间（单位:s）
)
);
$context = stream_context_create($options);
$result = file_get_contents($url, false, $context);
return $result;
}
}
$server = new Swoole();登录后复制
