php消息队列 推送模板消息,介绍swoole异步群发模板消息

1、用的是TP5.1的框架，swoole分成一个客户端发送接收消息，一个服务器负责处理信息

服务端代码，服务器要先安装swoole拓展，用 php server.php 启动进程监听

推荐(免费)：swoole<?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 thinkcachedriverRedis;

//use thinkController;

use thinkDb;

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} taskn";

$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} finishn";

}

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();

2、客户端请求，可以通过api访问function send_tem_to(){

$type = input('type'); // 0 按人头算 1 按标签算 2 全部粉丝

$target = input('target/s');

$field = input('fields/s');

$tem_id = input('tem_id');//模板ID，字符串

$url = input('url','');

$client = new swoole_client(SWOOLE_SOCK_TCP);//创建同步TCP

if (!$client->connect('127.0.0.1', 9501, 0.5))//链接

{

exit("connect failed. Error: {$client->errCode}n");

}

$client->send(json_encode(array('appid'=>$this->appid,'uid'=>$this->uid,'tem_id'=>$tem_id,'data'=>$field))); //发送请求

$rec = $client->recv();//接收返回数据

$client->close();//关闭链接

}