dev

admin/control/refill_order.php

@@ -1,7 +1,7 @@
 <?php
 
 require_once(BASE_HELPER_PATH . '/refill/util.php');
-require_once(BASE_HELPER_PATH . '/task/task.php');
+require_once(BASE_HELPER_PATH . '/task/task_helper.php');
 
 class refill_orderControl extends SystemControl
 {
@@ -168,9 +168,11 @@ class refill_orderControl extends SystemControl
         {
             $manager = new task\manager();
             $task = $manager->add_task('refill_order_stat',$cond,0);
-
-//            $stat = $this->all_order_state_stat($condition);
-//            Tpl::output('stat', $stat);
+            if($task->completed() && $task->success())
+            {
+                $stat = $task->result();
+                Tpl::output('stat', $stat);
+            }
         }
 
         $dispatcher_queue_length = refill\util::dispatcher_queue_length();

crontab/control/minutes.php

@@ -15,6 +15,7 @@ require_once(BASE_HELPER_PATH . '/refill/divert_account.php');
 require_once(BASE_HELPER_PATH . '/refill/RefillFactory.php');
 require_once(BASE_CORE_PATH . '/framework/function/http.php');
 require_once(BASE_HELPER_PATH . '/PHPExcel/PHPExcel.php');
+require_once(BASE_HELPER_PATH . '/task/task_helper.php');
 
 
 class minutesControl extends BaseCronControl
@@ -33,7 +34,6 @@ class minutesControl extends BaseCronControl
         $this->_fetch_order_unavaliable();
         $this->_check_merchant_alarm_amount();
 
-//        $this->_refill_task();
         Log::record(__FUNCTION__ . " end",Log::DEBUG);
 
 //        $this->_web_index_update();
@@ -410,13 +410,14 @@ class minutesControl extends BaseCronControl
     }
 
     //任务队列处理
-    public function onTaskOp()
+    public function taskOp()
     {
         $task_manager = new task\manager();
         $task_id = 0;
         while (true)
         {
-            try {
+            try
+            {
                 $task = $task_manager->pop_task($task_id);
                 if (!empty($task)) {
                     $task_manager->handle($task);

data/model/task.model.php

@@ -26,6 +26,7 @@ class taskModel extends Model
         return $this->where($condition)->find();
     }
 
+
     public function UnDispose(): array
     {
         $condition['is_del'] = 0;

docker/compose/stanley/master-cron/docker-compose.yml

@@ -33,4 +33,15 @@ services:
       - ../conf/etc/localtime:/etc/localtime:ro
       - ../conf/php/php-debug.ini:/usr/local/etc/php/php.ini
     container_name: "panda-accedit"
-    command: [php,"/var/www/html/crontab/index.php",'minutes','account_edit']
+    command: [php,"/var/www/html/crontab/index.php",'minutes','account_edit']
+
+  crontask:
+    image: php-zts-debug:7.3.18
+    volumes:
+      - ../../../../:/var/www/html
+      - /Volumes/Transcend/upload:/var/www/html/data/upload
+      - /Users/stanley-king/work/PHPProject/shoplog:/var/www/html/data/log
+      - ../conf/etc/localtime:/etc/localtime:ro
+      - ../conf/php/php-debug.ini:/usr/local/etc/php/php.ini
+    container_name: "panda-task"
+    command: [php,"/var/www/html/crontab/index.php",'minutes','task']

docker/compose/xyz/slave-crond/docker-compose.yml

@@ -11,4 +11,16 @@ services:
       - /nfs/upload:/var/www/html/data/upload
       - /mnt/shoplog:/var/www/html/data/log
     container_name: "panda-slavecron"
-    command: [crond,"-f"]
+    command: [crond,"-f"]
+
+  crontask:
+    image: php-zts-debug:7.3.18
+    volumes:
+      - ../../../../:/var/www/html
+      - ../conf/etc/localtime:/etc/localtime:ro
+      - ../conf/php/php.ini:/usr/local/etc/php/php.ini
+      - ../conf/crontab/slave_root:/var/spool/cron/crontabs/root
+      - /nfs/upload:/var/www/html/data/upload
+      - /mnt/shoplog:/var/www/html/data/log
+    container_name: "panda-task"
+    command: [php,"/var/www/html/crontab/index.php",'minutes','task']

helper/task/handler.php

@@ -3,39 +3,50 @@
 
 namespace task;
 
+use Exception;
 
 class handler
 {
     public function refill_order_stat($condition)
     {
-        $items = Model('')->table('refill_order,vr_order')
-            ->field('count(*) as order_count, sum(refill_amount) as refill_amounts, sum(channel_amount) as channel_amounts, sum(mch_amount) as mch_amounts, order_state')
-            ->join('inner')
-            ->on('refill_order.order_id=vr_order.order_id')
-            ->where($condition)
-            ->group('order_state')
-            ->select();
-
-        $all = [];
-        $data['order_count'] = $data['refill_amounts'] = $data['channel_amounts'] = $data['mch_amounts'] = 0;
-        $sending = $success = $cancel = $data;
-        foreach ($items as $item)
+        try
         {
-            if ($item['order_state'] == ORDER_STATE_SEND) {
-                $sending = $item;
-            } elseif ($item['order_state'] == ORDER_STATE_SUCCESS) {
-                $success = $item;
-            } elseif ($item['order_state'] == ORDER_STATE_CANCEL) {
-                $cancel = $item;
+            $items = Model('')->table('refill_order,vr_order')
+                ->field('count(*) as order_count, sum(refill_amount) as refill_amounts, sum(channel_amount) as channel_amounts, sum(mch_amount) as mch_amounts, order_state')
+                ->join('inner')
+                ->on('refill_order.order_id=vr_order.order_id')
+                ->where($condition)
+                ->group('order_state')
+                ->select();
+
+            $all = [];
+            $data['order_count'] = $data['refill_amounts'] = $data['channel_amounts'] = $data['mch_amounts'] = 0;
+            $sending = $success = $cancel = $data;
+            foreach ($items as $item)
+            {
+                if ($item['order_state'] == ORDER_STATE_SEND) {
+                    $sending = $item;
+                } elseif ($item['order_state'] == ORDER_STATE_SUCCESS) {
+                    $success = $item;
+                } elseif ($item['order_state'] == ORDER_STATE_CANCEL) {
+                    $cancel = $item;
+                }
+
+                $all['order_count'] += $item['order_count'];
+                $all['refill_amounts'] += ncPriceFormat($item['refill_amounts']);
+                $all['channel_amounts'] += ncPriceFormat($item['channel_amounts']);
+                $all['mch_amounts'] += ncPriceFormat($item['mch_amounts']);
             }
 
-            $all['order_count'] += $item['order_count'];
-            $all['refill_amounts'] += ncPriceFormat($item['refill_amounts']);
-            $all['channel_amounts'] += ncPriceFormat($item['channel_amounts']);
-            $all['mch_amounts'] += ncPriceFormat($item['mch_amounts']);
-        }
+            $result = ['all' => $all, 'sending' => $sending, 'success' => $success, 'cancel' => $cancel];
+
+            return [true,$result];
 
-        return ['all' => $all, 'sending' => $sending, 'success' => $success, 'cancel' => $cancel];
+        }
+        catch (Exception $ex)
+        {
+            return [false,false];
+        }
     }
 
     public function refill_order_stat_title($condition)

helper/task/manager.php

@@ -2,6 +2,9 @@
 
 namespace task;
 
+use Exception;
+use trans_wapper;
+
 class manager
 {
     private $mHandler;
@@ -16,6 +19,7 @@ class manager
             return false;
         }
 
+        //task 唯一性
         $hash_fun = function ($method,$params) {
             $val = $method . '-' . serialize($params);
             return md5($val);
@@ -26,24 +30,42 @@ class manager
         $title = $this->mHandler->$title_fun($params);
 
         $mod_task = Model('task');
-        $task = $mod_task->getByHash($task_hash);
-        if(empty($task))
+        $item = $mod_task->getByHash($task_hash);
+        if(empty($item))
         {
-            $task = ['type' => $method, 'title' => $title,
+            $item = ['type' => $method, 'title' => $title,
                      'params' => serialize($params), 'task_hash' => $task_hash,
                      'add_time' => time(), 'is_show' => $is_show];
-            $task_id = $mod_task->insert($task);
-            return $task_id;
-        }
-        else {
-            return $task;
+            $task_id = $mod_task->insert($item);
+            $item = $mod_task->where(['task_id' => $task_id])->find();
         }
+
+        return task_wrapper::CreateTask($item);
     }
 
-    //弹出一个任务，并且讲任务设为进行中
+    //弹出一个任务，并且将任务设为进行中
     public function pop_task($task_id)
     {
-        return ['method' => '',];
+        $mod_task = Model('task');
+        $trans = new trans_wapper($mod_task,'pop_task');
+
+        try {
+            $item = $mod_task->field('*')
+                             ->where(['task_id' => ['gt',$task_id],'state' => task_wrapper::Waiting])
+                             ->master(true)->lock(true)->find();
+
+            if(!empty($item)) {
+                $mod_task->where(['task_id' => $item['task_id']])->update(['state' => task_wrapper::Processing,'act_time' => time()]);
+            }
+            $trans->commit();
+
+            return $item;
+        }
+        catch (Exception $ex)
+        {
+            $trans->rollback();
+            return [];
+        }
     }
 
     //任务结束，并且
@@ -65,10 +87,19 @@ class manager
     public function handle($task)
     {
         $method = $task['type'];
-        $params = $task['params'];
+        $params = unserialize($task['params']);
 
         $handler = new handler();
         [$succ,$result] = $handler->$method($params);
 
+        $mod_task = Model('task');
+        if(!empty($result))
+        {
+            $mod_task->where(['task_id' => $task['task_id']])->update(['state' => task_wrapper::Handled,
+                'result_state' => $succ ? 1 : 0, 'result' => serialize($result),
+                'finish_time' => time()]);
+        }
+
+        return $task['task_id'];
     }
 }

helper/task/task.php

@@ -4,4 +4,4 @@ namespace task;
 
 require_once(BASE_HELPER_PATH . '/task/manager.php');
 require_once(BASE_HELPER_PATH . '/task/handler.php');
-
+require_once(BASE_HELPER_PATH . '/task/task_wrapper.php');

helper/task/task_wrapper.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace task;
+
+
+class task_wrapper
+{
+    //1待处理、2处理中、3已处理、4已经取消'
+    const Waiting = 1;
+    const Processing = 2;
+    const Handled = 3;
+    const Canceled = 4;
+
+    private $mRecord;
+    public function __construct($item)
+    {
+        $this->mRecord = $item;
+    }
+
+    public function completed() : bool
+    {
+        return $this->mRecord['state'] == self::Handled;
+    }
+
+    public function success() : bool {
+        return $this->mRecord['result_state'] == 1;
+    }
+
+    public function result()
+    {
+        if($this->success()) {
+            return unserialize($this->mRecord['result']);
+        } else {
+            return [];
+        }
+    }
+
+    public static function CreateTask($data)
+    {
+        return new task_wrapper($data);
+    }
+}