参考：[https://www.kancloud.cn/uuling/tp5-console/366131](https://www.kancloud.cn/uuling/tp5-console/366131)

```

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('结束');

}

}

```