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

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();//關閉連結

}