RedisQueueBatchHandler.php

<?php


namespace Meibuyu\Micro\Handler;
use Hyperf\Redis\Redis;

abstract class RedisQueueBatchHandler
{
    /**
     * 重试次数
     */
    protected $retry=0;

    protected $queue_name;

    /**
     * @var Redis
     */
    protected $redis;

    const MAX_RETRY_TIMES = 3;

    const BATCH_DEAL_NUM = 36;

    const ERROR_RETRY_CODE = 9000;

    /**
     * RedisQueueBatchHandler constructor.
     */
    public function __construct()
    {
        $this->queue_name = $this->specifyQueueName();
        $this->redis = $this->specifyRedisServer();
    }


    /**
     * @param $data
     * @param $specifyQueue
     */
    public function addToQueue($data,$specifyQueue='')
    {
        $this->specifyRedisServer()->rPush($specifyQueue?:$this->queue_name,json_encode($data));
    }

    /**
     * 将执行失败的数据重回队列
     * @param $arr
     */
    protected function backToQueue($arr)
    {
        array_unshift($arr,$this->queue_name);
        call_user_func_array([$this->redis,'lPush'],$arr);
    }

    //批处理具体逻辑
    abstract protected function batchDeal($data);
    abstract protected function specifyQueueName();
    abstract protected function specifyRedisServer() : Redis;


    public function consume()
    {

        while (true){
            //有数据则放回 否则延迟20s后返回空数组，同时可保持连接活跃
            //执行过程 如果redis服务异常，调用io操作时异常退出，框架重新拉起该进程
            $exist = $this->redis->blPop($this->queue_name,30);
            if(empty($exist)) continue ;
            $this->redis->lPush($this->queue_name,$exist[1]);

            //每次从列表取100
            $arr = $this->redis->lRange($this->queue_name,0,self::BATCH_DEAL_NUM-1);

            //取完 从队列删掉
            $this->redis->lTrim($this->queue_name,count($arr),-1);
            //数据格式化
            $formatArr = array_map(function ($item){
                return json_decode($item,true);
            },$arr);
            try {
                //具体批处理逻辑
                $this->batchDeal($formatArr);

            }catch (\Throwable $exception){

                //错误码为100 重新推到队列
                if($exception->getCode()==self::ERROR_RETRY_CODE){
                    if($this->retry<self::MAX_RETRY_TIMES){ //重试次数不超过3次
                        $this->backToQueue($arr);
                        $this->retry++;
                    }else{
                        $this->retry = 0; //重置当前次数
                        $this->errorWriteToFile($formatArr,$exception);
                    }

                }else{
                    $this->retry = 0;
                    $this->errorWriteToFile($formatArr,$exception);
                }

            }

        }

    }

    /**
     * @param $data
     * @param \Throwable $exception
     */
    private function errorWriteToFile($data, \Throwable $exception)
    {
        put_log(
            json_encode($data)."\n".
            $exception->getMessage()."\n".
            $exception->getFile().' line:'.
            $exception->getLine()."\n".
            $exception->getTraceAsString()

            ,'RedisQueue/'.$this->queue_name.'/'.today()
        );
    }


}