<?php
namespace Adapter\PlatformName;
use Domain\Datahub\Instance\Storage\DataStatus;
use Domain\Datahub\Instance\Adapter\Adapter;
use Domain\Datahub\Instance\LogMessage;
use Domain\Datahub\Instance\Storage\LogStatus;
class PlatformNameExecuteAdapter extends Adapter
{
const DIRECTION = 'target';
private $times = 0;
/**
* 调度方法
*
* @return void
*/
public function dispatch()
{
$this->times++;
if ($this->times >= 30) {
$this->asynTargetJobDispatch(10); // 重新激活 dt 命令
return;
}
$data = $this->getDataStorage()->fetch(); // 从mongodb 获取待处理数据
if (count($data) === 0) {
return $this->_returnDispatch();
}
$request = $this->generateRequestParams($data); // 转化原始数据为目标平台写入数据
$request = $this->removeNull($request);//清除空值
if (!$request) {
$this->getLogStorage()->insertOne(['text' => LogMessage::DISPATCH_TARGET_REQUEST_ERROR, 'request' => $request], LogStatus::ERROR);
$this->dispatch();
return;
}
// 标记原始数据为 队列中 插入到目标平台队列池
$jobId = $this->getAsynTargetJobStorage()->insertOne($this->metaData['api'], [$request], $this->getDataStorage()->ids, $this->getDataStorage()->dataRange);
$this->getDataStorage()->setFetchStatus(DataStatus::QUEUE, null, null, new \MongoDB\BSON\ObjectId($jobId));
$this->jobs[] = $jobId;
// 开始进行排队
$this->asynTargetJob(round($this->asynTimes), $jobId);
$this->asynTimes += 1.4;
$this->dispatch();
return true;
}
}
目标平台写入 $adapter->dispatch();
步骤1: 递增调度次数 $times++,超过最大调度次数排队 ??秒 下一次调度
步骤2: 从Mongodb取得一批【待处理】数据
步骤3: 数据+元数据生成请求参数
步骤4:写入任务到队列存储器返回jobid
步骤5:任务开始进行排队(自定义延迟)
步骤6:标记数据正在排队
步骤7:递归循环调度下一次
目标平台写入 $adapter->handleResponse();
public function handleResponse($response, $jobId = null)
{
$this->getLogStorage()->insertOne(['text' => 'handleResponse', 'response' => $response], LogStatus::RECORD);
if ($response['Success'] != true) {
return $this->handleError($response, $jobId);
}
$this->getAsynTargetJobStorage()->updateResponse($jobId, DataStatus::FINISHED, $response, [], null, $this->active);
$this->handleSuccessCallback($response, $jobId);
return $response;
}
错误日志方法
public function handleError($response, $jobId = null)
{
$throw = new PlatformThrowable($this);
$throw->handle($jobId, $response);
$this->getAsynTargetJobStorage()->updateResponse($jobId, DataStatus::ERROR, $response, [], null, $this->active);
$this->getLogStorage()->insertOne(['text' => LogMessage::INVOKE_FAIL, 'response' => $response], LogStatus::ERROR);
return $response;
}
目标平台写入 $adapter->connect();
同源平台方法一样。
2021-08-08 09:24:12 | |
2021-05-12 04:31:41 | |
2021-01-11 22:34:35 | |
2021-05-28 19:50:29 | |
2021-10-17 05:17:10 | |
2021-03-26 12:57:06 | |
2021-08-12 03:42:31 | |
2021-10-05 11:29:04 | |
2021-04-26 12:15:56 | |
2021-02-09 11:45:05 | |
2021-05-02 07:08:11 | |
2021-06-16 07:54:14 | |
2021-05-16 13:47:09 | |
2021-09-25 11:20:12 | |
2021-05-25 11:11:48 | |
2021-08-10 03:08:07 | |
2021-08-06 18:58:21 | |
2021-05-23 10:51:32 | |
2021-02-16 16:44:22 | |
2021-03-16 07:31:29 | |
2023-01-26 10:05:12 | |
2023-01-26 10:05:11 | |
2023-01-26 10:05:10 | |
2023-01-26 10:05:09 | |
2023-01-26 10:05:08 |
卢剑航 13760755942
数据集成专家 拥有十多年丰富的经验,擅长ERP、MES、数据中台、营销云中台等集成。他能够根据客户需求,为其提供一站式集成解决方案,帮助企业快速实现各类系统数据集成服务。
黄宏棵 13286997615
数据集成顾问 资深系统集成顾问,专长于ERP、电商OMS、钉钉及CRM系统。他能提供高效的集成方案,优化企业运营流程,提升业务效率和决策智能化。
何海波 18175716035
数据集成顾问 轻易云的技术专家,拥有丰富的数据集成规划经验。他能够为客户提供专业、全面的数据集成规划方案,熟练掌握多种集成技术和工具,帮助企业在数据集成领域得到长远发展。