变更

BackupCommand.php

@@ -5,6 +5,8 @@ namespace Database;
 use Co\Channel;
 use Exception;
 use Kiri\Di\LocalService;
+use Kiri\Error\StdoutLoggerInterface;
+use Swoole\Coroutine;
 use Swoole\Process;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -26,6 +28,8 @@ class BackupCommand extends Command
 	
 	private LocalService $service;
 	
+	public array $percentStatus = [];
+	
 	
 	/**
 	 *
@@ -70,15 +74,16 @@ class BackupCommand extends Command
 			$database = next($databaseInfo);
 			
 			$path = $input->getArgument('path');
+			if (!str_starts_with($path, '/')) {
+				$path = APP_PATH . $path;
+			}
 			
-			$data = $input->getOption('data');
+			$isData = $input->getOption('data');
-
 			if (!is_dir($path)) {
 				mkdir($path);
 			}
 			
-
+			$waite = new Coroutine\WaitGroup();
-			$processes = [];
 			foreach ($table as $value) {
 				$tableInfo = $data->createCommand('show create table `' . $data->database . '`.`' . $value . '`')->one();
 				
@@ -97,17 +102,20 @@ class BackupCommand extends Command
 				
 				file_put_contents($tmp, $tableCreator . ';' . PHP_EOL . PHP_EOL, FILE_APPEND);
 				
-				if ($data == 1) {
+				if ($isData == 1) {
-					$process = new Process(fn(Process $process) => $this->writeData($process, $database, $value), false,
+					$waite->add(1);
-						2, true);
+					Coroutine::create(function () use ($waite, $input, $value, $tmp) {
-					$process->start();
+						defer(function () use ($waite) {
+							$waite->done();
+						});
+						$this->writeData( $input->getOption('database'), $value, $tmp);
+					});
+				}
+			}
 			
-					$processes[] = $process;
+			$waite->wait();
-				}
+			
-			}
+			$output->write('dump data success');
-			foreach ($processes as $process) {
-				Process::wait();
-			}
 		} catch (\Throwable $throwable) {
 			$output->writeln($throwable->getMessage());
 		} finally {
@@ -116,23 +124,30 @@ class BackupCommand extends Command
 	}
 	
 	
+	public bool $isEnd = false;
+	
+	
 	/**
-	 * @param Process $process
 	 * @param string $dbname
-	 * @param string $value
+	 * @param string $tableName
+	 * @param string $path
 	 * @return void
 	 * @throws Exception
 	 */
-	public function writeData(Process $process, string $dbname, string $tableName, string $path): void
+	public function writeData(string $dbname, string $tableName, string $path): void
 	{
 		$offset = 0;
 		$size   = 1000;
 		
 		$channel = new Channel(200);
-		for ($i = 0; $i < $channel->length(); $i++) {
+		for ($i = 0; $i < 200; $i++) {
-			go(function () use ($channel, $path, $dbname, $tableName) {
+			Coroutine::create(function () use ($channel, $path, $dbname, $tableName) {
 				while ($channel->errCode != SWOOLE_CHANNEL_CLOSED) {
-					$value = $channel->pop();
+					$value = json_decode($channel->pop(), true);
+					
+					if ($value === null) {
+						continue;
+					}
 					
 					$value = $this->toSql($dbname, $tableName, $value);
 					
@@ -141,23 +156,59 @@ class BackupCommand extends Command
 			});
 		}
 		
+		$id = Coroutine::create(function () use ($channel) {
+			$data = Coroutine::waitSignal(SIGTERM | SIGINT);
+			if ($data) {
+				$this->isEnd = true;
+			}
+		});
+		
 		/** @var Connection $database */
 		$database = \Kiri::service()->get($dbname);
-		while (true) {
+		
+		$total = $database->createCommand("SELECT COUNT(*) as total FROM " . $tableName)->one()['total'];
+		
+		$startTime = time();
+		
+		$wait = new Coroutine\WaitGroup();
+		while ($this->isEnd === false && $offset < $total) {
+			$wait->add(1);
+			Coroutine::create(function () use ($wait, $offset, $size, $total, $dbname, $tableName, $channel) {
+				defer(function () use ($wait) {
+					$wait->done();
+				});
+				/** @var Connection $database */
+				$database = \Kiri::service()->get($dbname);
+				
 				$data = $database->createCommand("SELECT * FROM $tableName LIMIT $offset,$size")->all();
-			$channel->push($data);
+				if (count($data) < 1) {
-			if (count($data) < $size) {
+					return;
-				break;
 				}
+				$channel->push(json_encode($data));
+				
+				$this->percentStatus[$dbname . $tableName] = intval(round(($offset + $size) / $total, 2) * 100);
+				
+				$this->outputProgress(true);
+			});
 			$offset += $size;
 		}
 		
+		$wait->wait();
+		
+		echo 'use time ' . (time() - $startTime) . 's' . PHP_EOL;
+		
 		$channel->close();
 		
-		$process->exit(0);
+		Coroutine::cancel($id);
 	}
 	
 	
+	/**
+	 * @param string $dbname
+	 * @param string $value
+	 * @param array $data
+	 * @return string
+	 */
 	public function toSql(string $dbname, string $value, array $data): string
 	{
 		$strings = ['INSERT INTO ' . $dbname . '.' . $value];
@@ -168,10 +219,91 @@ class BackupCommand extends Command
 				$strings[] = '(' . implode(',', $keys) . ') VALUES';
 			} else {
 				$keys   = array_values($datum);
-				$strings[] = '(' . implode(',', $keys) . '),';
+				$encode = [];
+				foreach ($keys as $val) {
+					if (is_string($val)) {
+						$encode[] = '\'' . htmlentities($val) . '\'';
+					} else {
+						$encode[] = $val;
 					}
 				}
-		return implode('', $strings);
+				$strings[] = '(' . implode(',', $encode) . '),';
+			}
+		}
+		return rtrim(implode('', $strings), ',') . ';';
+	}
+	
+	
+	/**
+	 * @param $key
+	 * @param $percent
+	 * @return string
+	 * 组合成进度条
+	 */
+	public function buildLine($key, $percent): string
+	{
+		$repeatTimes = 100;
+		if ($percent > 100) {
+			$percent = 100;
+		}
+		if ($percent > 0) {
+			$hasColor = str_repeat('■', $percent);
+		} else {
+			$hasColor = '';
+		}
+		
+		if ($repeatTimes - $percent > 0) {
+			$noColor = str_repeat(' ', $repeatTimes - $percent);
+		} else {
+			$noColor = '';
+		}
+		
+		$buffer = "[{$hasColor}{$noColor}]";
+		if ($percent !== 100) {
+			$percentString = sprintf("[  %s %-6s]", $key, $percent . '%');
+		} else {
+			$percentString = sprintf("[  %s %-5s]", $key, 'OK');;
+		}
+		
+		return $percentString . $buffer . "\r";
+	}
+	
+	/**
+	 * @param bool $clear
+	 * @return void
+	 * 输出进度条
+	 */
+	public function outputProgress(bool $clear = false): void
+	{
+		if ($clear) {
+			$number = count($this->percentStatus);
+			for ($i = 0; $i < $number; $i++) {
+				system("tput cuu1");
+				system("tput el");
+			}
+		}
+		foreach ($this->percentStatus as $key => $value) {
+			echo $this->buildLine($key, $value) . "\n";
+		}
+	}
+	
+	/**
+	 * @param $k
+	 * @param $value
+	 * @return void
+	 * 更新进度条值
+	 */
+	public function updateProgressValue($k, $value): void
+	{
+		$this->percentStatus[$k] = $value;
+		if ($this->percentStatus[$k] >= 1000) {
+			$this->percentStatus[$k] = 1000;
+			$this->outputProgress(true);
+			return;
+		}
+		
+		$this->outputProgress(true);
+		usleep(50000);
 	}
 	
 }