log = Zc::getLog(); $this->checkLog = Zc::getLog('timer/check/revive_queue', ZcLog::INFO); $this->timerTriggerLog = Zc::getLog('timer/timer_trigger', ZcLog::INFO); $this->tcLog = Zc::getLog('timer/tc', ZcLog::INFO); $this->db = Zc::getDb(); $this->timerDao = Zc::singleton('TimerDao'); $this->adminOperateLogDao = Zc::singleton(AdminBackgroundLogDao::class); $this->timerRedis = RedisExt::factory('timerCache'); } /** * 尝试去获取Timerlock,能得到就继续;否则就结束 * * @return boolean number */ public function tryGetTimerLock($timerCode, $hostname, $timerLockId, $suspTimeout = 180, $sureTimeout = 600, $useDbId) { $timerLock = $this->getTimerLock($timerLockId, $useDbId); if ((strpos($hostname, '-pre') === false) && Zc::C('env') == 'live' && sys_getloadavg()[0] >= 25 && (strpos($timerCode, 'move') === false || strpos($timerCode, 'buffer') === false)) { $maxCount = 0; } else { $timerSpec = $this->getTimerSpec($timerCode, $hostname, $useDbId); $maxCount = (int)$timerSpec['max_count']; } $count = $this->getTimerLockCount($timerCode, $hostname, $suspTimeout, $sureTimeout, $useDbId); $this->tcLog->info('tryGetTimerLock getTimerLockCount count ' . $count . ', maxCount ' . $maxCount); if ($count === false) { return false; } $targetTimerLockId = false; if ($count < $maxCount) { if (!empty($timerLock)) { $targetTimerLockId = $timerLock['timer_lock_id']; } else { $targetTimerLockId = $this->addTimerLock($timerCode, $hostname, $useDbId); $this->tcLog->info('tryGetTimerLock addTimerLock ID: ' . $targetTimerLockId); } } else if (($maxCount > 0) && ($count == $maxCount)) { if (!empty($timerLock)) { $targetTimerLockId = $timerLock['timer_lock_id']; } } else { if (!empty($timerLock)) { $this->tcLog->info('tryGetTimerLock need deleteTimerLock'); $this->deleteTimerLock($timerLock['timer_lock_id'], $useDbId, $timerCode, $hostname); // $targetTimerLockId = $timerLock['timer_lock_id']; 超出了,本次就结束了吧。 } } if ($targetTimerLockId) { $this->timerDao->setTimerCodeLastActiveTime($timerCode, date('Y-m-d H:i:s')); } //$this->tcLog->info('tryGetTimerLock ret ' . ZcArrayHelper::sp($targetTimerLockId)); return $targetTimerLockId; } private function buildRedisKeyTimerSpecTimerCodeHostname($timerCode, $hostname, $useDbId) { return $useDbId . '_ts_' . $hostname . '_' . $timerCode; } private function getTimerSpec($timerCode, $hostname, $useDbId) { $rk = $this->buildRedisKeyTimerSpecTimerCodeHostname($timerCode, $hostname, $useDbId); //$this->tcLog->info($rk); $cacheRow = unserialize($this->timerRedis->get($rk)); //$this->tcLog->info($cacheRow); if (!empty($cacheRow)){ $this->tcLog->info("get rk {$rk} cache sucess count is " . count($cacheRow)); return $cacheRow; } $row = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_code = %s and `hostname` = %s', $timerCode, $hostname); //$this->tcLog->info($row); if (!empty($row)) { $r = $this->timerRedis->setex($rk, $this->timerDao->getTimerSpecCacheExpireTime(), serialize($row)); $this->tcLog->info('setex ts' . $r); } //$this->tcLog->info($row); return $row; } private function buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId) { return $this->timerDao->buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId); } private function getTimerLockCount($timerCode, $hostname, $suspTimeout, $sureTimeout, $useDbId) { $rows = $this->getHostnameTimerLockList($hostname, $useDbId, $timerCode); if ($rows === false) { return false; } if (empty($rows)) { $this->tcLog->info('getTimerLockCount::getTimerLockList empty rows'); return 0; } $count = 0; foreach ($rows as $ret) { if ((time() - strtotime($ret['gmt_last_heartbeat'])) > $suspTimeout) { continue; } $count++; } foreach ($rows as $ret) { if ((time() - strtotime($ret['gmt_last_heartbeat'])) > $sureTimeout) { $this->deleteTimerLock($ret['timer_lock_id'], $useDbId, $timerCode, $hostname); } } return $count; } private function getHostnameTimerLockList($hostname, $useDbId, $timerCode) { $rk = $this->buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId); $cacheMap = $this->timerRedis->hGetAll($rk); //$this->tcLog->info("getHostnameTimerLockList $rk hGetAll: " . print_r($cacheMap, true)); if (!empty($cacheMap)) { foreach ($cacheMap as $key => $value){ $rows[$key] = unserialize($value); } } if (empty($rows)) { $rows = $this->db->useDbIdOnce($useDbId)->query('select * from timer_lock where timer_code = %s and `hostname` = %s', $timerCode, $hostname); if ($rows === false) { return false; } $rows = ZcArrayHelper::changeKeyRow($rows, 'timer_lock_id'); //$this->tcLog->info('getHostnameTimerLockList::getTimerLockList From DB: ' . print_r($rows, true)); if(!empty($rows)){ $hRowsMap = array(); foreach ($rows as $timerLockId => $row) { $hRowsMap[$timerLockId] = serialize($row); } $r1 = $this->timerRedis->hMset($rk, $hRowsMap); $r2 = $this->timerRedis->expire($rk, $this->timerDao->getTimerInfoCacheExpireTime()); //$this->tcLog->info('getHostnameTimerLockList saveTimerLockList ret' . print_r($hRowsMap, true) . ", r1 $r1, r2 $r2"); } } $timerLockIds = array_keys($rows); $heartbeatMap = $this->timerDao->mgetTimerLockLastHeartbeatTimeFromCache($timerLockIds, $useDbId); foreach ($heartbeatMap as $timerLockId => $gmtLastHeartbeat) { $rows[$timerLockId]['gmt_last_heartbeat'] = $gmtLastHeartbeat; } return $rows; } private function getTimerLock($timerLockId, $useDbId) { if (empty($timerLockId)) { return array(); } $rk = $this->timerDao->buildRedisKeyTimerLockTimerLockId($timerLockId, $useDbId); $catchRow = unserialize($this->timerRedis->get($rk)); if (!empty($catchRow)) { return $catchRow; } $row = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_lock where timer_lock_id = %i', $timerLockId); if (!empty($row)){ $this->timerRedis->setex($rk, $this->timerDao->getTimerInfoCacheExpireTime(), serialize($row)); } return $row; } private function addTimerLock($timerCode, $hostname, $useDbId) { $now = date("Y-m-d H:i:s"); $insert = array ( 'timer_code' => $timerCode, 'hostname' => $hostname, 'gmt_last_heartbeat' => $now, 'gmt_create' => $now, 'gmt_modified' => $now, ); $ret = $this->db->useDbIdOnce($useDbId)->insert('timer_lock', $insert); if ($ret) { $rk = $this->buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId); $this->timerRedis->del($rk); $this->tcLog->info('addTimerLock then del ' . $rk); } return ($ret === false) ? false : $this->db->lastInsertId(); } public function getTimerSpecList($filter, $pageNo, $pageSize) { $wheres = array(); if ($filter['timer_code']) { $wheres[] = $this->db->prepare('and `timer_code` like %ss', $filter['timer_code']); } if ($filter['hostname']) { $wheres[] = $this->db->prepare('and `hostname` like %ss', $filter['hostname']); } $where = empty($wheres) ? '' : implode(' ', $wheres); $dbIdMap = ['timer']; if (!empty($filter['dbId']) && in_array($filter['dbId'], $dbIdMap)) { $dbIdMap = [$filter['dbId']]; } $rows = array(); $maxTotal = 0; foreach ($dbIdMap as $dbId) { list($timerRows, $total) = $this->db->useDbIdOnce($dbId)->queryPage("SELECT * FROM `timer_spec` WHERE `timer_code` IS NOT NULL %l ORDER BY `timer_spec_id` desc", $where, $pageNo, $pageSize); if (empty($timerRows)) { continue; } foreach ($timerRows as $k => $row) { $timerRows[$k]['db_id'] = $dbId; } $rows = array_merge($rows, $timerRows); $maxTotal = ($total > $maxTotal) ? $total : $maxTotal; } return array($rows, $maxTotal); } public function getTimerLockList($filter, $pageNo, $pageSize) { $wheres = array(); if ($filter['timer_code']) { $wheres[] = $this->db->prepare('and `timer_code` like %ss', $filter['timer_code']); } if ($filter['hostname']) { $wheres[] = $this->db->prepare('and `hostname` like %ss', $filter['hostname']); } $dbId = 'timer'; $where = empty($wheres) ? '' : implode(' ', $wheres); return $this->db->useDbIdOnce($dbId)->queryPage("SELECT * FROM `timer_lock` WHERE `timer_code` IS NOT NULL %l ORDER BY `timer_code` ASC, `gmt_last_heartbeat` DESC", $where, $pageNo, $pageSize); } public function updateTimerLockHeartbeat($timerLockId, $useDbId) { $this->timerDao->updateTimerLockHeartbeat($timerLockId, null, $useDbId); } public function deleteTimerLock($timerLockId, $useDbId, $timerCode, $hostname) { return $this->timerDao->deleteTimerLock($timerLockId, $useDbId, $timerCode, $hostname); } public function getNeedReviveTimerLockList($useDbId = 'timer') { $timerLockList = $this->db->useDbIdOnce($useDbId)->query("SELECT tl.timer_lock_id, tl.timer_code, tl.hostname, tl.gmt_last_heartbeat, ti.queue_name FROM timer_lock tl INNER JOIN timer_info ti ON tl.timer_code = ti.timer_code WHERE ti.`status` = %s and ti.timeout > 0 AND tl.gmt_last_heartbeat < DATE_SUB(NOW(), INTERVAL ti.timeout SECOND)", StatusConst::normal); return $this->mergeReviveTimerHeartbeat($timerLockList, $useDbId); } private function mergeReviveTimerHeartbeat($timerLockList, $useDbId){ if (empty($timerLockList)) { return false; } $timerLockList = ZcArrayHelper::changeKeyRow($timerLockList, 'timer_lock_id'); $timerLockIds = array_keys($timerLockList); $chunks = array_chunk($timerLockIds, 50, true); foreach ($chunks as $chunk) { $heartbeatMap = $this->timerDao->mgetTimerLockLastHeartbeatTimeFromCache($chunk, $useDbId); foreach ($heartbeatMap as $timerLockId => $gmtLastHeartbeat) { $timerLockList[$timerLockId]['gmt_last_heartbeat'] = $gmtLastHeartbeat; } } return $timerLockList; } public function unlockQueueByLockId($timerLockInfo, $lockId, $useDbId = 'timer') { $queueName = $timerLockInfo['queue_name']; $hostname = $timerLockInfo['hostname']; $timerCode = $timerLockInfo['timer_code']; $oldErrorMode = $this->db->setErrorMode(ZcDb::ERROR_MODE_EXCEPTION); $transStatus = $this->db->startTransaction(null, $useDbId); $update = array( 'locked' => 0, 'gmt_modified' => ZcDbEval::now() ); $where = array(); switch ($queueName) { case 'batch_task' : $where[] = $this->db->prepare('AND `status` IN %ls', array(StatusConst::wait, StatusConst::waitRetry, StatusConst::waitRevert, StatusConst::waitRevert, StatusConst::processingCancel)); break; case 'image_delete_task' : $where[] = $this->db->prepare('AND `status` IN %ls', array(StatusConst::processing)); $update['status'] = StatusConst::wait; break; } $whereString = implode(' ', $where); try { $dr = $this->deleteTimerLock($lockId, $useDbId, $timerCode, $hostname); $this->checkLog->info($this->db->lastSql() . ' -> ' . $dr); list($useDbId, ) = DbRoute::getDbAndTbl($queueName); $ur = $this->db->useDbIdOnce($useDbId)->update($queueName, $update, 'locked = %i %l', $lockId, $whereString); $this->checkLog->info($this->db->lastSql() . ' -> ' . $ur); $this->db->commit($transStatus); $this->db->setErrorMode($oldErrorMode); return true; } catch (Exception $ex) { $this->checkLog->error($ex->getMessage()); $this->db->rollback($transStatus); $this->db->setErrorMode($oldErrorMode); return false; } } public function getActiveTimerSpecList($hostname, $dbIds) { $timerSpecList = array(); foreach ($dbIds as $dbId) { $smallTimeSpecList = $this->getActiveTimerSpecListByDbId($hostname, $dbId); $timerSpecList = array_merge($timerSpecList, $smallTimeSpecList); } return $timerSpecList; } private function getActiveTimerSpecListByDbId($hostname, $dbId) { $now = ZcDateHelper::now(); $timerSpecList = $this->db->useDbIdOnce($dbId)->query('select timer_spec_id,timer_code,wait_sleep_second,next_sleep_second,cron_expression,gmt_next_run,driver from timer_spec where hostname = %s and max_count > 0 and `status` = %s', $hostname, StatusConst::normal); foreach ($timerSpecList as $k => &$timerSpec) { if ($timerSpec['driver'] == 'python') { unset($timerSpecList[$k]); continue; } if (!is_numeric($timerSpec['wait_sleep_second'])) { $timerSpec['wait_sleep_second'] = 15; } if (!is_numeric($timerSpec['next_sleep_second'])) { $timerSpec['next_sleep_second'] = 0.2; } $timerSpec['cron_expression'] = trim($timerSpec['cron_expression']); if (!empty($timerSpec['cron_expression'])) { $cron = Cron\CronExpression::factory($timerSpec['cron_expression']); $gmtNextRun = $cron->getNextRunDate()->format('Y-m-d H:i:s'); if (empty($timerSpec['gmt_next_run']) || !$cron->isDue($timerSpec['gmt_next_run'])) { $r = $this->db->useDbIdOnce($dbId)->exec('update timer_spec set gmt_next_run = %s where timer_spec_id = %i', $gmtNextRun, $timerSpec['timer_spec_id']); $this->timerTriggerLog->info("nullOrNotDue $dbId, so update gmtNextRun to [$gmtNextRun], ret[$r]. " . ZcArrayHelper::sp($timerSpec)); unset($timerSpecList[$k]); continue; } if ($now < $timerSpec['gmt_next_run']) { $this->timerTriggerLog->info("untimely $dbId: " . ZcArrayHelper::sp($timerSpec)); unset($timerSpecList[$k]); continue; } $r = $this->db->useDbIdOnce($dbId)->exec('update timer_spec set gmt_prev_run = %s, gmt_next_run = %s where timer_spec_id = %i', $now, $gmtNextRun, $timerSpec['timer_spec_id']); $this->timerTriggerLog->info("timely $dbId, update gmtNextRun to [$gmtNextRun], ret[$r]. " . ZcArrayHelper::sp($timerSpec)); } } $timerSpecList = array_values($timerSpecList); return $timerSpecList; } public function searchTimerMonitorTblLogList($filter, $page, $pageSize) { $where = array (); if ($filter['keyword']) { $where[] = $this->db->prepare('AND (tml.monitor_name like %ss OR tm.`desc` like %ss)', $filter['keyword'], $filter['keyword']); } if (!empty($filter['startDt'])) { $where[] = $this->db->prepare("and tml.gmt_check >= %s", date('Y-m-d H:i:s', strtotime($filter['startDt']))); } if (!empty($filter['endDt'])) { $where[] = $this->db->prepare("and tml.gmt_check <= %s", date('Y-m-d H:i:s', strtotime($filter['endDt']))); } if (!empty($filter['monitorType'])) { $where[] = $this->db->prepare('AND tml.monitor_type = %s', $filter['monitorType']); } if (!empty($filter['group'])) { $where[] = $this->db->prepare('AND tm.group = %s', $filter['group']); } $whereString = implode(' ', $where); list($logDbId, $realMonitorTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor); list($logDbId, $realMonitorLogTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor_log); list ($monitorLogList, $total) = $this->db->useDbIdOnce($logDbId)->queryPage(" SELECT tml.*, tm.`desc`, tm.group FROM %b tml LEFT JOIN %b tm ON tm.monitor_name = tml.monitor_name AND tm.monitor_type = tml.monitor_type WHERE TRUE %l ORDER BY SUBSTR(tml.gmt_check, 1, 16) DESC, tml.total_num DESC, tml.monitor_name ASC", $realMonitorLogTbl, $realMonitorTbl, $whereString, $page, $pageSize); return array ( $monitorLogList, $total ); } public function getAllTimerMonitorLogList($monitorName, $monitorType, $startDt, $endDt) { list($logDbId, $realMonitorLogTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor_log); return $this->db->useDbIdOnce($logDbId)->query("SELECT total_num, user_total, gmt_check FROM %b WHERE monitor_name = %s AND monitor_type = %s AND gmt_check >= %s AND gmt_check <= %s ORDER BY gmt_check ASC", $realMonitorLogTbl, $monitorName, $monitorType, $startDt, $endDt); } public function getTimerMonitorInfo($timerMonitorlId) { list($logDbId, $realMonitorTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor); return $this->db->useDbIdOnce($logDbId)->queryFirstRow("SELECT * FROM %b WHERE timer_monitor_id = %i", $realMonitorTbl, $timerMonitorlId); } public function getTimerMonitorList($filter) { $where = array(); if ($filter['keyword']) { $where[] = $this->db->prepare('AND (monitor_name like %ss OR `desc` like %ss)', $filter['keyword'], $filter['keyword']); } if ($filter['monitorType']) { $where[] = $this->db->prepare('AND (monitor_type = %s )', $filter['monitorType']); } if ($filter['status']) { $where[] = $this->db->prepare('AND `status` = %s', $filter['status']); } if ($filter['group']) { $where[] = $this->db->prepare('AND `group` = %s', $filter['group']); } $whereString = implode(' ', $where); list($logDbId, $realMonitorTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor); return $this->db->useDbIdOnce($logDbId)->query("SELECT * FROM %b WHERE TRUE %l ORDER BY monitor_name", $realMonitorTbl, $whereString); } public function updateTimerMonitor($timerMonitorlId, $data) { $update = ZcArrayHelper::filterColumns($data, array('monitor_name', 'desc', 'group', 'monitor_type', 'primary_key_id', 'status', 'gmt_check')); $update['gmt_modified'] = ZcDbEval::now(); list($logDbId, $realMonitorTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor); return $this->db->useDbIdOnce($logDbId)->update($realMonitorTbl, $update, 'timer_monitor_id = %i', $timerMonitorlId); } public function addTimerMonitor($data) { $insert = ZcArrayHelper::filterColumns($data, array('monitor_name', 'desc', 'group', 'monitor_type', 'primary_key_id', 'status')); $insert['gmt_create'] = ZcDbEval::now(); $insert['gmt_modified'] = ZcDbEval::now(); list($logDbId, $realMonitorTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor); $ret = $this->db->useDbIdOnce($logDbId)->insert($realMonitorTbl, $insert); if ($ret) { return $this->db->lastInsertId(); } return false; } public function addTimerMonitorLog($data) { $insert = ZcArrayHelper::filterColumns($data, array('monitor_name', 'monitor_type', 'total_num', 'user_total', 'max_gmt_create', 'min_gmt_create', 'gmt_check')); $insert['gmt_modified'] = ZcDbEval::now(); $insert['gmt_create'] = ZcDbEval::now(); list($logDbId, $realMonitorLogTbl) = DbRoute::getDbAndTbl(TblConst::timer_monitor_log); $ret = $this->db->useDbIdOnce($logDbId)->insert($realMonitorLogTbl, $insert); if ($ret) { return $this->db->lastInsertId(); } return false; } public function getMonitorBizData($monitorTable, $checkDt, $primaryKeyId, $filter = array()) { $wheres = array(); if (isset($primaryKeyId)) { $queueIdColumnName = QueueConst::getQueueIdColumnName($monitorTable); $wheres[] = $this->db->prepare('and %b > %i', $queueIdColumnName, $primaryKeyId); } $whereString = implode(' ', $wheres); $startDt = date("Y-m-d 00:00:00", strtotime($checkDt) - 86400); $endDt = date("Y-m-d 23:59:59", strtotime($checkDt) - 86400); list($realDb, $realTbl) = DbRoute::getDbAndTbl($monitorTable); if (isset($primaryKeyId)) { $statList = $this->db->useDbIdOnce($realDb)->query("SELECT COUNT(*) user_total, MAX(%b) max_primary_key_id, MIN(gmt_create) min_gmt_create, MAX(gmt_create) max_gmt_create FROM %b WHERE gmt_create >= %s AND gmt_create <= %s %l GROUP BY user_id", $queueIdColumnName, $realTbl, $startDt, $endDt, $whereString); } else { $statList = $this->db->useDbIdOnce($realDb)->query("SELECT COUNT(*) user_total, MIN(gmt_create) min_gmt_create, MAX(gmt_create) max_gmt_create FROM %b WHERE gmt_create >= %s AND gmt_create <= %s %l GROUP BY user_id", $realTbl, $startDt, $endDt, $whereString); } $totalNum = 0; $userTotal = 0; $minGmtCreate = null; $maxGmtCreate = null; $maxPrimaryKeyId = isset($primaryKeyId) ? $primaryKeyId : null; $gmtChecke = ($checkDt < ZcDateHelper::today()) ? date('Y-m-d H:i:s', strtotime($checkDt)) : date('Y-m-d H:i:s'); foreach ($statList as $statDetail) { $userTotal++; $totalNum += $statDetail['user_total']; $minGmtCreate = (!empty($minGmtCreate) && ($minGmtCreate < $statDetail['min_gmt_create'])) ? $minGmtCreate : $statDetail['min_gmt_create']; $maxGmtCreate = (!empty($maxGmtCreate) && ($maxGmtCreate > $statDetail['max_gmt_create'])) ? $maxGmtCreate : $statDetail['max_gmt_create']; if (isset($primaryKeyId)) { $maxPrimaryKeyId = (!empty($maxPrimaryKeyId) && ($maxPrimaryKeyId > $statDetail['max_primary_key_id'])) ? $maxPrimaryKeyId : $statDetail['max_primary_key_id']; } } return array ( 'total_num' => $totalNum, 'user_total' => $userTotal, 'max_gmt_create' => $maxGmtCreate, 'min_gmt_create' => $minGmtCreate, 'gmt_check' => $gmtChecke, 'primary_key_id' => $maxPrimaryKeyId, ); } public function getSmsResponseQueueStat() { $statDetail = $this->db->queryFirstRow("SELECT COUNT(*) total, MIN(gmt_create) min_gmt_create, MAX(gmt_create) max_gmt_create FROM sms_response WHERE verify = 0"); $gmtChecke = date('Y-m-d H:i:s'); $totalNum = $statDetail['total']; $userTotal = 0; $minGmtCreate = $statDetail['min_gmt_create']; $maxGmtCreate = $statDetail['max_gmt_create']; return array ( 'total_num' => $totalNum, 'user_total' => $userTotal, 'max_gmt_create' => $maxGmtCreate, 'min_gmt_create' => $minGmtCreate, 'gmt_check' => $gmtChecke ); } public function getMonitorQueueData($monitorTable) { list($realDb, $realTbl) = DbRoute::getDbAndTbl($monitorTable); if ($realTbl == 'ware_detect_task_queue') { $realDb = 'zc.slave1'; } $statList = $this->db->useDbIdOnce($realDb)->query("SELECT user_id, COUNT(*) user_total, MIN(gmt_create) min_gmt_create, MAX(gmt_create) max_gmt_create FROM %b WHERE TRUE %l GROUP BY user_id", $realTbl); $gmtChecke = date('Y-m-d H:i:s'); $totalNum = 0; $userTotal = 0; $minGmtCreate = null; $maxGmtCreate = null; foreach ($statList as $statDetail) { $userTotal++; $totalNum += $statDetail['user_total']; $minGmtCreate = (!empty($minGmtCreate) && ($minGmtCreate < $statDetail['min_gmt_create'])) ? $minGmtCreate : $statDetail['min_gmt_create']; $maxGmtCreate = (!empty($maxGmtCreate) && ($maxGmtCreate > $statDetail['max_gmt_create'])) ? $maxGmtCreate : $statDetail['max_gmt_create']; } return array ( 'total_num' => $totalNum, 'user_total' => $userTotal, 'max_gmt_create' => $maxGmtCreate, 'min_gmt_create' => $minGmtCreate, 'gmt_check' => $gmtChecke ); } private function delTimerSpecCache($timerSpecId, $useDbId) { $data = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_spec_id = %i', $timerSpecId); if (empty($data)) { return false; } //删除redis timer spec 缓存 $rk = $this->buildRedisKeyTimerSpecTimerCodeHostname($data['timer_code'], $data['hostname'], $useDbId); $this->timerRedis->del($rk); } public function updateTimerSpec($timerSpecId, $data, $useDbId, $adminId) { $row = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_spec_id = %i', $timerSpecId); if (empty($row)) { return CommonTool::failResult("数据不存在"); } $dbIdMap = array_keys(Zc::C(ZcConfigConst::DbConfig)['connections']); if (!in_array($useDbId, $dbIdMap)) { return CommonTool::failResult("DbId 错误"); } $update = ZcArrayHelper::filterColumns($data, ['timer_code', 'hostname', 'status', 'max_count', 'wait_sleep_second', 'next_sleep_second', 'cron_expression', 'driver']); if (empty($update)) { return CommonTool::failResult("更新数据为空"); } $this->delTimerSpecCache($timerSpecId, $useDbId); $update = array_map('trim', $update); if (empty($update['wait_sleep_second'])) { $update['wait_sleep_second'] = null; } if (empty($update['next_sleep_second'])) { $update['next_sleep_second'] = null; } if (empty($update['cron_expression'])) { $update['cron_expression'] = null; } $update['gmt_modified'] = ZcDbEval::now(); $affRow = $this->db->useDbIdOnce($useDbId)->update('timer_spec', $update, 'timer_spec_id = %i', $timerSpecId); $logContent = "更新定时器,影响行数[$affRow]," . json_encode($update, JSON_UNESCAPED_UNICODE); $this->adminOperateLogDao->add($adminId, AdminBackgroundConst::actionUpdateTimerSpec, $logContent); //删除redis缓存 if (($update['status'] == StatusConst::pause) || ($update['max_count'] == 0)) { $row = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_spec_id = %i', $timerSpecId); $timerCode = $row['timer_code']; $this->resetTimerInfoByTimerCode($timerCode, $useDbId); } return CommonTool::successResult(array( 'timer_spec_id' => $timerSpecId )); } public function deleteTimerSpec($useDbId, $timerSpecId, $adminId) { $row = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_spec_id = %i', $timerSpecId); if (empty($row)) { return CommonTool::failResult("数据不存在"); } $affRow = $this->db->useDbIdOnce($useDbId)->delete('timer_spec', 'timer_spec_id = %i', $timerSpecId); if ($affRow === false) { return CommonTool::failResult("数据删除错误"); } $logContent = "删除定时器,timerCode[{$row['timer_code']}],host[{$row['hostname']}],影响行数[{$affRow}]"; $this->adminOperateLogDao->add($adminId, AdminBackgroundConst::actionDeleteTimerSpec, $logContent); //删除redis缓存 $timerCode = $row['timer_code']; $this->resetTimerInfoByTimerCode($timerCode, $useDbId); $this->delTimerSpecCache($timerSpecId, $useDbId); return CommonTool::successResult(array( 'timer_spec_id' => $timerSpecId )); } public function updateTimerSpecNum($num, $timerCode) { list($timerDbId, $timerSpecTbl) = DbRoute::getDbAndTbl(TblConst::timer_spec); return $this->db->useDbIdOnce($timerDbId)->update($timerSpecTbl, [ 'max_count' => $num ], 'timer_code = %s ', $timerCode); } private function resetTimerInfoByTimerCode($timerCode, $useDbId){ $hostnames = $this->db->useDbIdOnce($useDbId)->queryFirstColumn('select distinct hostname from timer_lock where timer_code = %s', $timerCode); foreach ($hostnames as $hostname) { $rk = $this->buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId); $this->timerRedis->del($rk); $rk = $this->buildRedisKeyTimerLockTimerCodeHostname($timerCode, $hostname, $useDbId); $this->timerRedis->del($rk); } } public function addTimerSpec($data, $useDbId, $adminId) { $dbIdMap = array_keys(Zc::C(ZcConfigConst::DbConfig)['connections']); if (!in_array($useDbId, $dbIdMap)) { return CommonTool::failResult("DbId 错误"); } $insert = ZcArrayHelper::filterColumns($data, ['timer_code', 'hostname', 'status', 'max_count', 'wait_sleep_second', 'next_sleep_second', 'cron_expression', 'driver']); if (empty($insert)) { return CommonTool::failResult("新增数据为空"); } $insert = array_map('trim', $insert); if (empty($insert['wait_sleep_second'])) { $insert['wait_sleep_second'] = null; } if (empty($insert['next_sleep_second'])) { $insert['next_sleep_second'] = null; } if (empty($insert['cron_expression'])) { $insert['cron_expression'] = null; } $affRow = $this->db->useDbIdOnce($useDbId)->queryFirstRow('select * from timer_spec where timer_code = %s and `hostname` = %s', $data['timer_code'], $data['hostname']); if (!empty($affRow)) { return CommonTool::failResult("定时器已存在,请检查输入是否错误"); } $insert['gmt_modified'] = ZcDbEval::now(); $insert['gmt_create'] = ZcDbEval::now(); $affRow = $this->db->useDbIdOnce($useDbId)->insert('timer_spec', $insert); if ($affRow === false) { return CommonTool::failResult("定时器添加失败"); } $logContent = "添加定时器,影响行数[$affRow]," . json_encode($data, JSON_UNESCAPED_UNICODE); $this->adminOperateLogDao->add($adminId, AdminBackgroundConst::actionAddTimerSpec, $logContent); //删除redis timer spec 缓存 $rk = $this->buildRedisKeyTimerSpecTimerCodeHostname($data['timer_code'], $data['hostname'], $useDbId); $this->timerRedis->del($rk); return CommonTool::successResult('timer_spec_id', $this->db->lastInsertId()); } public function downTimerSpecMaxCount() { $aff = $this->db->useDbIdOnce('timer')->exec("update timer_spec set max_count_backup = max_count, max_count = 0 where max_count > 0"); return $aff; } public function upTimerSpecMaxCount() { $af = $this->db->useDbIdOnce('timer')->exec("update timer_spec set max_count = max_count_backup where max_count_backup > 0 and max_count = 0"); return $af; } public function updateQueueLastHeartbeat($queueName, $queueId, $timerLockId = null) { return $this->timerDao->updateQueueLastHeartbeat($queueName, $queueId, $timerLockId); } public function heartbeat() { return $this->timerDao->heartbeat(); } }