异步协程与异步日志批处理

63ec3025 · Liu lu · 937681f1 · 63ec3025
Commit 63ec3025 authored Feb 12, 2022 by Liu lu
Hide whitespace changes
Inline Side-by-side

Showing with 237 additions and 0 deletions

README.md README.md +237 -0

No files found.
--- a/README.md
+++ b/README.md
@@ -153,3 +153,240 @@ human_time(time()-strtotime('2020-06-06 12:12'));
 info('aaa',[1,2,3],new stdClass(){$a=1;},collect([1,23,4]));
 info(1);
 ```
+
+### 5、数据表批量操作
+
+用法：
+
+   继承 \Meibuyu\Micro\Model\BaseModel 的模型：
+```
+    class LogTrace extends BaseModel
+    {
+    
+        protected $table = 'trace_logs';
+    
+        /**
+         * 是否使用时间戳管理
+         * @var bool
+         */
+        public $timestamps = false;
+    
+        /**
+        * 可写入数据的字段.
+        * @var array
+        */
+        protected $fillable = [
+    		'source',
+    		'origin_params',
+    		'is_completed',
+    		'process_info',
+    
+        ];
+     }
+```
+
+得到基于主键或唯一索引作为条件的俩个批处理方法： 
+```
+    //批量更新,、$data为二维数组，必须包含主键或唯一索引的数据
+    //参数二缺省为主键名，或唯一索引名
+    LogTrace::getModel()->batchUpdateByField($data,'request_id')
+    
+    //基于ON DUPLICATE KEY UPDATE 批量更新或插入 $data 必须包含主键或唯一索引的数据
+    LogTrace::getModel()->batchUpdateOrCreateByUniqueKey($data);
+```
+
+### 6、基于@LogTrace()注解，实现异步日志队列服务
+用法：
+#### 1）、建立日志跟踪表
+```sql
+CREATE TABLE `trace_logs` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `request_id` varchar(32) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '一次http或rpc请求调用的唯一key',
+  `source` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '来源，包含调用类命名空间及方法',
+  `origin_params` json NOT NULL COMMENT '记录注解方法被调用开始的原始传参',
+  `is_completed` tinyint(1) NOT NULL DEFAULT '0' COMMENT '此请求是否完成，使用LogTraceHandler::markComplete()标记',
+  `process_info` mediumtext COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '执行过程中输出，使用LogTraceHandler::recordProcess()记录',
+  `created_at` datetime NOT NULL COMMENT '日志记录开始时间',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `request_id` (`request_id`)
+) ENGINE=InnoDB AUTO_INCREMENT=5868 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+```
+#### 2）、定义消费进程，日志批量更新到数据库
+```php
+<?php
+/**
+ * 异步日志队列批处理
+ */
+namespace App\Process;
+
+use Hyperf\Process\AbstractProcess;
+use Hyperf\Process\Annotation\Process;
+use Meibuyu\Micro\Handler\LogTrace\LogTraceQueue;
+
+/**
+ * @Process(name="SyncTraceLog")
+ */
+class SyncTraceLog extends AbstractProcess
+{
+
+    /**
+     * 进程数量
+     * @var int
+     */
+    public $nums = 1;
+    /**
+     * 进程名称
+     * @var string
+     */
+    public $name = 'syn-trace-log';
+
+    /**
+     * 管道类型
+     * @var int
+     */
+    public $pipeType = 2;
+
+    /**
+     * 是否启用协程
+     * @var bool
+     */
+    public $enableCoroutine = true;
+
+
+    /**
+     * @inheritDoc
+     */
+    public function handle(): void
+    {
+        make(LogTraceQueue::class)->consume();
+    }
+}
+```
+#### 3）、对操作方法指定注解，主动记录日志信息
+
+```
+  给test方法加上 @LogTrace() 注解，从此处开始记录日志，可在此请求的任何流程地方
+  使用Meibuyu\Micro\Handler\LogTrace\LogTraceHandler::recordProcess
+  手动记录输出，下面只是最简单的示例
+
+    /**
+     * @LogTrace() 
+     * @return string
+     * @throws \Exception
+     */
+    public function test()
+    {
+          try {
+              //记录数组
+              LogTraceHandler::recordProcess($this->request->all());
+
+              //流程1
+              LogTraceHandler::recordProcess('执行到流程1');
+
+              //流程2
+              LogTraceHandler::recordProcess('执行到流程2');
+
+              //流程3 抛出一个异常
+              throw new Exception('test111');
+
+              //流程执行完成标记结束
+              LogTraceHandler::markComplete();
+
+          }catch (\Throwable $exception){
+              //记录异常日志
+              LogTraceHandler::recordProcess($exception);
+          }
+          return  'test222';
+    }
+
+
+
+   ##执行过程输出到 trace_logs表 process_info:
+   
+        array (
+          'scanNo' => 'SPUS-20211202-158-3',
+        )
+        
+        执行到流程1
+        
+        执行到流程2
+        
+        抛出一个异常
+        /var/www/runtime/container/proxy/App_Controller_IndexController.proxy.php line:80
+        #0 /var/www/vendor/hyperf/di/src/Aop/ProceedingJoinPoint.php(84): App\Controller\IndexController->App\Controller\{closure}()
+        #1 /var/www/vendor/hyperf/di/src/Aop/ProxyTrait.php(85): Hyperf\Di\Aop\ProceedingJoinPoint->processOriginalMethod()
+        #2 /var/www/vendor/hyperf/utils/src/Pipeline.php(104): App\Controller\IndexController::Hyperf\Di\Aop\{closure}()
+        #3 /var/www/vendor/hyperf/di/src/Aop/ProceedingJoinPoint.php(69): Hyperf\Utils\Pipeline::Hyperf\Utils\{closure}()
+        #4 /var/www/vendor/meibuyu/micro/src/Aspect/LogTraceAspect.php(32): Hyperf\Di\Aop\ProceedingJoinPoint->process()
+        #5 /var/www/vendor/hyperf/di/src/Aop/Pipeline.php(30): Meibuyu\Micro\Aspect\LogTraceAspect->process()
+        #6 /var/www/vendor/hyperf/utils/src/Pipeline.php(95): Hyperf\Di\Aop\Pipeline->Hyperf\Di\Aop\{closure}()
+        #7 /var/www/vendor/hyperf/di/src/Aop/ProxyTrait.php(86): Hyperf\Utils\Pipeline->then()
+        #8 /var/www/vendor/hyperf/di/src/Aop/ProxyTrait.php(29): App\Controller\IndexController::handleAround()
+        #9 /var/www/runtime/container/proxy/App_Controller_IndexController.proxy.php(88): App\Controller\IndexController::__proxyCall()
+        #10 /var/www/vendor/hyperf/http-server/src/CoreMiddleware.php(161): App\Controller\IndexController->test1()
+        #11 /var/www/vendor/hyperf/http-server/src/CoreMiddleware.php(113): Hyperf\HttpServer\CoreMiddleware->handleFound()
+        #12 /var/www/vendor/hyperf/dispatcher/src/AbstractRequestHandler.php(64): Hyperf\HttpServer\CoreMiddleware->process()
+        #13 /var/www/vendor/hyperf/dispatcher/src/HttpRequestHandler.php(26): Hyperf\Dispatcher\AbstractRequestHandler->handleRequest()
+        #14 /var/www/vendor/hyperf/dispatcher/src/HttpDispatcher.php(40): Hyperf\Dispatcher\HttpRequestHandler->handle()
+        #15 /var/www/vendor/hyperf/http-server/src/Server.php(116): Hyperf\Dispatcher\HttpDispatcher->dispatch()
+        #16 {main}
+        
+        返回结果："test222"
+```
+####使用说明
+> 1. 对方法加上 @LogTrace() 注解，建议注解的地方为http请求和rpc调用的入口处，便于使用脚本拿到原始传参便捷发起重试
+> 2. 使用@LogTrace()注解的方法逻辑内任意地方使用LogTraceHandler::recordProcess记录输出（如须异步协程里也跟踪，第二参数须为true）
+> 3. 日志跟踪数据都存放在trace_logs表，每一次请求或rpc调用都对应一条唯一记录，process_info字段按顺序记录了流程输出
+
+
+
+### 7、基于@AsyncCoroutine()注解，对方法实现异步协程处理
+
+对于耗时长，加快影响效率，可以使用写队列，消费处理。
+也可以将这部分逻辑放到异步协程去处理，用法：
+
+```
+    http 请求test1，调用 延迟5s的continueTest（该方法已加入AsyncCoroutine注解），
+    但接口马上返回结果。5s后，后台将continueTest方法逻辑执行输出到控制台或者写入日志
+
+
+      /**
+       * @LogTrace()
+       * @return array
+       * @throws \Exception
+       */
+      public function test1()
+      {
+          //此处调用异步协程去处理，立刻返回结果
+          $this->continueTest($this->request->all());
+  
+          return Coroutine::id();
+  
+      }
+  
+      /**
+       * 使用AsyncCoroutine注解，使该方法投递到子协程里执行
+       * @AsyncCoroutine() 
+       */
+      private function continueTest($params)
+      {
+          sleep(5); //睡眠5s 
+          LogTraceHandler::recordProcess(Coroutine::id(),true);
+          LogTraceHandler::recordProcess(Coroutine::parentId(),true);
+  
+          return Coroutine::id();
+      }
+
+
+```
+##### 使用须知
+```
+  1. 给某个方法加上异步协程AsyncCoroutine注解，该方法被投放到另一个协程执行，
+  该方法的传参尽量不要使用模型对象（等连接资源对象），如果使用模型对象投递到
+  另一个协程，进行更新操作，会造成数据错乱。如要使用应禁止更新等操作
+
+  2. 配合基于@LogTrace()注解异步日志队列服务，可以使用 LogTraceHandler::recordProcess
+    对异步协程执行的情况进行跟踪，但在异步协程里，第二个参数必须为true
+    如  LogTraceHandler::recordProcess('记录输出数据',true);
+    
+```
\ No newline at end of file