swoole異步群發模板消息

1、用的是TP5.1的框架，swoole分成一個客戶端發送接收消息，一個服務器負責處理信息

　　服務端代碼，服務器要先安裝swoole拓展，用 php server.php 啟動進程監聽

　　

<?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();//關閉鏈接
        
    }