执行数据库Xa事务
@access public
@param  callable $callback 数据操作方法回调
@param  array    $dbs      多个查询对象或者连接对象
@return mixed
@throws PDOException
@throws \Exception
@throws \Throwable
178502
public function transactionXa($callback, array $dbs = [])
    {
        $xid = uniqid('xa');

        if (empty($dbs)) {
            $dbs[] = $this->getConnection();
        }

        foreach ($dbs as $key => $db) {
            if ($db instanceof Query) {
                $db = $db->getConnection();

                $dbs[$key] = $db;
            }

            $db->startTransXa($xid);
        }

        try {
            $result = null;
            if (is_callable($callback)) {
                $result = call_user_func_array($callback, [$this]);
            }

            foreach ($dbs as $db) {
                $db->prepareXa($xid);
            }

            foreach ($dbs as $db) {
                $db->commitXa($xid);
            }

            return $result;
        } catch (\Exception $e) {
            foreach ($dbs as $db) {
                $db->rollbackXa($xid);
            }
            throw $e;
        } catch (\Throwable $e) {
            foreach ($dbs as $db) {
                $db->rollbackXa($xid);
            }
            throw $e;
        }
    }
491790
public function tryUntil($callback)
    {
        try {
            return call_user_func_array($this->process, $this->args);
        } catch (\Exception $e) {
            if (in_array(get_class($e), $this->allowedExceptions)) {
                throw $e;
            }
        }

        $this->logger->warning(sprintf('[%s] Waiting until callback return a success', $this->logInfos['callable']));

        $startTime = time();
        $retry = 1;

        for (;;) {
            if ($callback instanceof YoloInterface) {
                $available = $callback->isAvailable();
            } else {
                $available = call_user_func($callback);
            }

            if (true === $available) {
                try {
                    return call_user_func_array($this->process, $this->args);
                } catch (\Exception $e) {
                    $this->logger->warning(sprintf('[%s] Can\'t perform action [%s/%s]', $this->logInfos['callable'], $retry, $this->maxRetry));

                    if (in_array(get_class($e), $this->allowedExceptions)) {
                        throw $e;
                    }

                    ++$retry;
                }

                if ($retry >= $this->maxRetry) {
                    $this->logger->error(sprintf('[%s] retry has failed, aborting [%s/%s]', $this->logInfos['callable'], $retry, $this->maxRetry));

                    $retryException = new YoloException('', 0, $e);
                    $retryException->setRetry($this->maxRetry);
                    $retryException->setTimeout($this->timeout);
                    throw $retryException;
                }
            } else {
                $this->logger->warning(sprintf('[%s] Can\'t perform action, waiting next tick', $this->logInfos['callable']));
            }

            if ($this->timeout > 0 && (time() - $startTime) >= $this->timeout) {
                $retryException = new YoloException('Timed out');
                $retryException->setRetry($this->maxRetry);
                $retryException->setTimeout($this->timeout);

                throw $retryException;
            }

            sleep(1);
        }
    }
336772
public function doCall($callback, $params = [], $fallback = null)
    {
        $data = $this->circuitBreaker->fallback();

        App::trace($this->getServiceName() . "服务，当前[开启状态]，执行服务fallback服务降级容错处理");
        // 开启定时器
        $nowTime = time();

        if ($this->circuitBreaker->isOpen()
            && $nowTime > $this->circuitBreaker->getSwitchOpenToHalfOpenTime()
        ) {
            $delayTime = $this->circuitBreaker->getDelaySwitchTimer();

            // 定时器不是严格的，新增3s,容错时间
            $switchToHalfStateTime = $nowTime + ($delayTime / 1000) + 3;
            App::getTimer()->addAfterTimer('openState', $delayTime, [$this, 'delayCallback']);
            $this->circuitBreaker->setSwitchOpenToHalfOpenTime($switchToHalfStateTime);

            App::trace($this->getServiceName() . "服务，当前[开启状态]，创建延迟触发器，一段时间后状态切换为半开状态");
        }

        return $data;
    }
330064
public function query($callback, $bindId = null, $sql = null, Context $context = null)
    {
        if ($sql == null) {
            $sql = $this->dbQueryBuilder->getStatement(false);
            $this->dbQueryBuilder->clear();
        }
        if (empty($sql)) {
            throw new Exception('sql empty');
        }
        $data = [
            'sql' => $sql
        ];
        $data['token'] = $this->addTokenCallback($callback);
        if (! empty($bindId)) {
            $data['bind_id'] = $bindId;
        }
        // 写入管道
        $this->asynManager->writePipe($this, $data, $this->workerId);
    }
266209
public function run(callable $callback, ...$params)
    {
        $item = array_shift($this->instances);
        if (empty($item)) {
            return call_user_func_array($callback, $params);
        }
        return $item->handle($callback, function () use ($callback, $params) {
            return $this->run($callback, ...$params);
        });
    }
408276
public function trans($callback)
    {
        if (is_callable($callback))
        {
            $this->getAdapter()->beginTransaction();
            
            try {
                $result = $callback($this);
                
                if ($result === false) {
                    $this->getAdapter()->rollback();
                } else {
                    $this->getAdapter()->commit();
                }
            } catch (\Exception $e) {
                $this->getAdapter()->rollback();
                throw $e;
            }
        }
        else
        {
            return false;
        }
    }
141471
public function findCallback(Closure $callback, $columns = ['*'])
    {
        return $this->postProcess( parent::findCallback($callback, $columns) );
    }
426454
function run(callable $callback = null) {
        if (isset($this->res))
            return null;
        $this->res = proc_open($this->cmd, $this->descriptor_spec, $this->pipes, $this->cwd, $this->env);
        if ($this->res) {
            $this->stdout = new TcpConnection($this->pipes[1]);
            if (is_callable($this->onMessage))
                $this->stdout->onMessage = function ($conn, $data) {
                    ($this->onMessage)($this, $data);
                };
            if (is_callable($this->onClose))
                $this->stdout->onClose = function () {
                    ($this->onClose)($this);
                };
            stream_set_blocking($this->pipes[0], false);
            if (is_callable($callback))
                $callback($this);
            return true;
        }
        unset ($this->res);
        return false;
    }
313790
public function execute($sqlStatement, $bindParams=null, $bindTypes=null) {
		try {
			return $this->callParentExecute($sqlStatement, $bindParams, $bindTypes);
		} catch(\PDOException $e) {
			if($e->getCode() === self::MYSQL_WENT_AWAY_CODE || $e->getMessage() === 'SQLSTATE[HY000]: General error: 2006 MySQL server has gone away') {
				$this->connect($this->_descriptor);
				return $this->callParentExecute($sqlStatement, $bindParams, $bindTypes);
			} else {
				throw $e;
			}
		}
	}
505995
public function asyncDnsLookup($callBack, array $headers = [])
    {
        $this->urlData['callBack'] = $callBack;
        $this->urlData['headers']  = $headers;

        $ip = self::getDnsCache($this->urlData['host']);
        if ($ip !== null) {
            $this->dnsLookupCallBack($ip);
        } else {
            $requestId = $this->getContext()->getRequestId();
            swoole_async_dns_lookup($this->urlData['host'], function ($host, $ip) use ($requestId) {
                if ($ip === '127.0.0.0') { // fix bug
                    $ip = '127.0.0.1';
                }

                if (empty(getInstance()->scheduler->taskMap[$requestId])) {
                    return;
                }

                if (empty($ip)) {
                    $this->getContext()->getLog()->warning($this->urlData['url'] . ' DNS查询失败');
                } else {
                    self::setDnsCache($host, $ip);
                    $this->dnsLookupCallBack($ip);
                }
            });
        }
    }
691685
public function call(callable $callback, $connection = null)
    {
        $this->database->connection($connection)->beginTransaction();

        try
        {
            $result = $callback($this->database);

            $this->database->connection($connection)->commit();
            $this->state = static::STATE_COMMIT;

            return $result;
        }
        catch (\Exception $exception)
        {
            $this->database->connection($connection)->rollBack();
            $this->state = static::STATE_ROLLBACK;

            return null;
        }
    }
175388
private function triggerCallback($callback, $arguments)
    {
        if (is_array($callback)) {
            $this->import($callback[0]);
            return call_user_func_array(array($this->{$callback[0]}, $callback[1]), $arguments);
        } elseif (is_callable($callback)) {
            return call_user_func_array($callback, $arguments);
        }

        return null;
    }
48133
public function setCallback($callback = null)
    {
        if (!is_null($callback) && !is_callable($callback)) {
            throw new Exception("callback must be of type callable or null");
        }

        $this->callback = $callback;

        return $this;
    }
595507
private function _exec($sql, $arguments = array())
    {
        $sql = $this->replaceArgs($sql, $arguments);

        // executes the statement
        $result = $this->_conn->query($sql);
        if ($this->_conn->errno > 0) {
            throw new DbException(
                "Failed to execute the statement: " .
                "({$this->_conn->errno}) {$this->_conn->error}"
            );
        }

        return $result;
    }
325779
public static function sync(bbn\db $db, $dbs='', $dbs_table='', $num_try = 0){

    if ( !$num_try ){
      self::def($dbs, $dbs_table);
      self::first_call();
      self::disable();
      $mode_db = self::$db->get_error_mode();
      $mode_dbs = self::$dbs->get_error_mode();
      self::$db->set_error_mode("continue");
      self::$dbs->set_error_mode("continue");
    }

    $num_try++;

    $to_log = [
      'deleted_sync' => 0,
      'deleted_real' => 0,
      'updated_sync' => 0,
      'updated_real' => 0,
      'inserted_sync' => 0,
      'inserted_real' => 0,
      'num_problems' => 0,
      'problems' => []
    ];


    $retry = false;

    $start = ( $test = self::$dbs->get_one("
      SELECT MIN(chrono)
      FROM ".self::$dbs->escape(self::$dbs_table)."
      WHERE db NOT LIKE ?
      AND state = 0",
      self::$db->current) ) ? $test : time();
    // Deleting the entries prior to this sync we produced and have been seen by the twin process
    $to_log['deleted_sync'] = self::$dbs->delete(self::$dbs_table, [
      ['db', '=', self::$db->current],
      ['state', '=', 1],
      ['chrono', '<', $start]
    ]);

    // Selecting the entries inserted
    $ds = self::$dbs->rselect_all(self::$dbs_table, ['id', 'tab', 'vals', 'chrono'], [
      ['db', '!=', self::$db->current],
      ['state', '=', 0],
      ['action', '=', 'insert']
    ], [
      'chrono' => 'ASC',
      'id' => 'ASC'
    ]);
    // They just have to be inserted
    foreach ( $ds as $i => $d ){
      if ( isset(self::$methods['cbf1']) ){
        self::cbf1($d);
      }
      $vals = \bbn\x::json_base64_decode($d['vals']);
      if ( !\is_array($vals) ){
        $to_log['num_problems']++;
        $to_log['problems'][] = "Hey, look urgently at the row $d[id]!";
      }
      else if ( self::$db->insert($d['tab'], $vals) ){
        if ( isset(self::$methods['cbf2']) ){
          self::cbf2($d);
        }
        $to_log['inserted_sync']++;
        self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
      }
      else if ( self::$db->select($d['tab'], [], $vals) ){
        self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
      }
      else{
        if ( $num_try > self::$max_retry ){
          $to_log['num_problems']++;
          $to_log['problems'][] = "Problem while syncing (insert), check data with status 5 and ID ".$d['id'];
          self::$dbs->update(self::$dbs_table, ["state" => 5], ["id" => $d['id']]);
        }
        $retry = 1;
      }
    }


    // Selecting the entries modified and deleted in the twin DB,
    // ordered by table and rows (so the same go together)
    $ds = self::$dbs->rselect_all(self::$dbs_table, ['id', 'tab', 'action', 'rows', 'vals', 'chrono'], [
      ['db', '!=', self::$db->current],
      ['state', '=', 0],
      ['rows', '!=', '[]'],
      ['action', '!=', 'insert']
    ], [
      'tab' => 'ASC',
      'rows' => 'ASC',
      'chrono' => 'ASC',
      'id' => 'ASC'
    ]);
    foreach ( $ds as $i => $d ){
      // Executing the first callback
      $d['rows'] = bbn\x::json_base64_decode($d['rows']);
      $d['vals'] = bbn\x::json_base64_decode($d['vals']);
      if ( isset(self::$methods['cbf1']) ){
        self::cbf1($d);
      }
      // Proceeding to the actions: delete is before
      if ( strtolower($d['action']) === 'delete' ){
        if ( self::$db->delete($d['tab'], $d['rows']) ){
          self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
          $to_log['deleted_real']++;
        }
        else if ( !self::$db->select($d['tab'], [], $d['rows']) ){
          self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
        }
        else{
          if ( $num_try > self::$max_retry ){
            self::$dbs->update(self::$dbs_table, ["state" => 5], ["id" => $d['id']]);
            $to_log['num_problems']++;
            $to_log['problems'][] = "Problem while syncing (delete), check data with status 5 and ID ".$d['id'];
          }
          $retry = 1;
        }
      }
      // Checking if there is another change done to this record and when in the twin DB
      $next_time = (
              isset($ds[$i+1]) &&
              ($ds[$i+1]['tab'] === $d['tab']) &&
              ($ds[$i+1]['rows'] === $d['rows'])
            ) ? $ds[$i+1]['chrono'] : microtime();
      // Looking for the actions done on this specific record in our database
      // between the twin change and the next (or now if there is no other change)
      $each = self::$dbs->rselect_all(self::$dbs_table, ['id', 'chrono', 'action', 'vals'], [
        ['db', '=', self::$db->current],
        ['tab', '=', $d['tab']],
        ['rows', '=', $d['rows']],
        ['chrono', '>=', $d['chrono']],
        ['chrono', '<', $next_time],
      ]);
      if ( \count($each) > 0 ){
        $to_log['num_problems']++;
        $to_log['problems'][] = "Conflict!";
        $to_log['problems'][] = $d;
        foreach ( $each as $e ){
          // If it's deleted locally and updated on the twin we restore
          if ( strtolower($e['action']) === 'delete' ){
            if ( strtolower($d['action']) === 'update' ){
              if ( !self::$db->insert_update(
                      $d['tab'],
                      bbn\x::merge_arrays(
                        $e['vals'],
                        $d['vals']
                      ))
              ){
                $to_log['num_problems']++;
                $to_log['problems'][] = "insert_update number 1 had a problem";
              }
            }
          }
          // If it's updated locally and deleted in the twin we restore
          else if ( strtolower($e['action']) === 'update' ){
            if ( strtolower($d['action']) === 'delete' ){
              if ( !self::$db->insert_update($d['tab'], bbn\x::merge_arrays($d['vals'], $e['vals'])) ){
                $to_log['num_problems']++;
                $to_log['problems'][] = "insert_update had a problem";
              }
            }
          // If it's updated locally and in the twin we merge the values for the update
            else if ( strtolower($d['action']) === 'update' ){
              $d['vals'] = bbn\x::merge_arrays($d['vals'], $e['vals']);
            }
          }
        }
      }
      // Proceeding to the actions update is after in case we needed to restore
      if ( strtolower($d['action']) === 'update' ){
        if ( self::$db->update($d['tab'], $d['vals'], $d['rows']) ){
          self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
          $to_log['updated_real']++;
        }
        else if ( self::$db->select($d['tab'], [], bbn\x::merge_arrays($d['rows'], $d['vals'])) ){
          self::$dbs->update(self::$dbs_table, ["state" => 1], ["id" => $d['id']]);
        }
        else{
          if ( $num_try > self::$max_retry ){
            self::$dbs->update(self::$dbs_table, ["state" => 5], ["id" => $d['id']]);
            $to_log['num_problems']++;
            $to_log['problems'][] = "Problem while syncing (update), check data with status 5 and ID ".$d['id'];
          }
          $retry = 1;
        }
      }
      // Callback number 2
      if ( isset(self::$methods['cbf2']) ){
        self::cbf2($d);
      }
    }


    $res = [];
    foreach ( $to_log as $k => $v ){
      if ( !empty($v) ){
        $res[$k] = $v;
      }
    }
    if ( $retry && ( $num_try <= self::$max_retry ) ){
      $res = bbn\x::merge_arrays($res, self::sync($db, $dbs, $dbs_table, $num_try));
    }
    else{
      self::$db->set_error_mode($mode_db);
      self::$dbs->set_error_mode($mode_dbs);
      self::enable();
    }
    return $res;
  }
617125
public function open($callback = null)
    {
        StatementManager::clear($this->db);
        $this->db->open();
        return $this->db->isConnected();
    }
988567
protected function run($query, $bindings, Closure $callback)
    {
        $this->reconnectIfMissingConnection();

        $start = microtime(true);

        // Here we will run this query. If an exception occurs we'll determine if it was
        // caused by a connection that has been lost. If that is the cause, we'll try
        // to re-establish connection and re-run the query with a fresh connection.
        try {
            $result = $this->runQueryCallback($query, $bindings, $callback);
        } catch (QueryException $e) {
            $result = $this->handleQueryException(
                $e, $query, $bindings, $callback
            );
            logger()->error('SQL Error', [$query, $bindings]);
            mstat()->set(1, 'BUG错误', 'SQL执行错误', $query, $e->getMessage(), 100);
        }

        $execute_time = $this->getElapsedTime($start);
        //监控SQL执行效率
        if ($execute_time) {
            mstat()->set(1, 'SQL执行效率', mstat()->formatTime($execute_time), $query);
        }

        // Once we have run the query we will calculate the time that it took to run and
        // then log the query, bindings, and execution time so we will report them on
        // the event that the developer needs them. We'll log time in milliseconds.
        $this->logQuery(
            $query, $bindings, $execute_time
        );

        return $result;
    }