think命令行 · 學習筆記

參考：[http://www.hmoore.net/uuling/tp5-console/366131](http://www.hmoore.net/uuling/tp5-console/366131) ``` <?php namespace app\command\consumer\order; use think\console\Command; use think\console\Input; use think\console\Output; use app\common\tool\Rabbit; use app\platform\logic\Daraz\Order; use app\platform\model\PlatformOrder; class DarazList extends Command { protected $input = null; protected $output = null; protected $colorList = [ 'red'=> "255;0;0;", 'orange'=> "255;165;0;", 'yellow'=> "255;255;0;", 'green'=> "0;255;0;", 'cyan'=> "0;127;255;", 'blue'=> "0;0;255;", 'purple'=> "139;0;255;", ]; protected function info($msg, $color="green"){ # ｛\033[｝為標簽起始｛m｝為標簽結束，屬性用數字加分號隔開 38是設置顏色 $colorRGB = isset($this->colorList[$color]) ? $this->colorList[$color] : $this->colorList['green']; $this->output->writeln("\033[38;2;".$colorRGB."1m".$msg."\033[0m"); } protected function configure() { // 指令配置 $this->setName('DarazList'); // 設置參數 } protected function execute(Input $input, Output $output) { $this->input = $input; $this->output = $output; // 指令輸出 $this->info('消費者程序啟動...'); $this->info('進程pid:' . getmypid()); $queueName = 'daraz_order_list'; dd($queueName); # 處理失敗的隊列名 $failedQueueName = 'failed_'.$queueName; $rabbit = Rabbit::instance('rabbit_mq_sync_task'); # 測試獲取單條隊列消息 // $data = $rabbit->consumeData($queueName, 5); // if(!empty($data)){ // // foreach($data as $d){ // // print_r(json_decode(gzuncompress($d['rawData']), true)); // // } // // $rabbit->ack($data['delivery_tag']); // dd(json_decode(($data[0]['rawData']), true)); // // dd($data); // } // $this->info('結束'); // die; # 消費計數 $i = 1; $point = 1000; $data = []; $delivery_tag = null; # 聲明對隊列消息的業務處理回調 $callback = function($msg) use (&$i, $point, &$data, &$delivery_tag){ if($i%$point == 0){ # 每處理指定數量的隊列消息可以標記下邏輯 } $this->info('第'.$i.'個消息處理'); $delivery_tag = $msg->delivery_info['delivery_tag']; $rawData = $msg->body; // $this->info('rawData:'.$i.' '. $rawData .PHP_EOL); # 業務邏輯處理 $data = json_decode($rawData, true); unset($rawData); # 必要參數校驗 $accountId = $data['accountId']; $platformCode = "VO"; $order = $data['order']; // dd($order); $orderLogic = new Order(); # 格式化數據 $formatData = $orderLogic->formatOrder($accountId, $platformCode, $order); // dd($formatData); // # 入庫 $orderModel = new PlatformOrder(); $result = $orderModel->syncOrder($formatData); # 如果有返回且插入失敗，要拋出插入異常 if(!$result){ throw new Exception("PlatformOrder->syncOrder failed", -1); } // $msg->delivery_info['channel']->basic_ack($delivery_tag); $i++; # 業務邏輯處理完，手動確認刪除，否則會一直繼續獲取隊列消息，并標記fetched unacked 只有消費腳本退出進程，才會釋放，但不會刪除 }; while(true){ try { $rabbit->consume($callback, $queueName, 0); }catch(Throwable $e){ # 如果是因為連接關閉，自動重連 if(strpos($e->getMessage(), 'Broken pipe or closed connection') !== false){ $this->info('連接斷開,開始重連'); $rabbit->close(); $rabbit->connect(); } # 異常, 統一丟到失敗隊列 $this->info('異常記錄：'.$e->getMessage()); dd("退出"); $rabbit->ack($delivery_tag); $failedMsg = [ 'raw_queue_name'=> $queueName, 'err_msg'=> $e->getMessage(), 'data'=> $data ]; $failedMsg = json_encode($failedMsg); $rabbit->easyProduce($failedMsg, $failedQueueName); $i++; continue; } } # 常駐進程退出 $this->info('結束'); } } ```