diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index 37198209..4cff90ac 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -14,6 +14,7 @@ use RdKafka\Exception;
 use RdKafka\KafkaConsumer;
 use RdKafka\TopicConf;
 use Server\Abstracts\CustomProcess;
+use Swoole\Coroutine;
 use Swoole\Process;
 use Throwable;
 
@@ -73,8 +74,7 @@ class Kafka extends CustomProcess
 
 			$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
 			do {
-				if ($this->checkProcessIsStop()) {
-					$this->exit();
+				if ($this->isStop) {
 					break;
 				}
 				$this->resolve($topic, $conf['interval'] ?? 1000);
@@ -94,18 +94,18 @@ class Kafka extends CustomProcess
 	{
 		try {
 			$message = $topic->consume(0, $interval);
-			if (empty($message)) {
-				return;
-			}
-			if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
-				$this->handlerExecute($message->topic_name, $message);
-			} else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
-				logger()->warning('No more messages; will wait for more');
-			} else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
-				logger()->error('Kafka Timed out');
-			} else {
-				logger()->error($message->errstr());
+			if (!empty($message)) {
+				if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
+					$this->handlerExecute($message->topic_name, $message);
+				} else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
+					logger()->warning('No more messages; will wait for more');
+				} else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
+					logger()->error('Kafka Timed out');
+				} else {
+					logger()->error($message->errstr());
+				}
 			}
+			Coroutine::sleep(0.01);
 		} catch (Throwable $exception) {
 			logger()->addError($exception, 'throwable');
 		}
diff --git a/Server/Abstracts/CustomProcess.php b/Server/Abstracts/CustomProcess.php
index cd88e86c..df073b67 100644
--- a/Server/Abstracts/CustomProcess.php
+++ b/Server/Abstracts/CustomProcess.php
@@ -57,8 +57,9 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
 			Coroutine::create(function () use ($process) {
 				/** @var Coroutine\Socket $message */
 				$message = $process->exportSocket();
-				error($message->recv());
-				$process->exit(0);
+				if ($message->recv() == 0x03455343213212) {
+					$this->waiteExit($process);
+				}
 			});
 			Coroutine::create(function () use ($process) {
 				$data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1);
@@ -68,7 +69,7 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
 						foreach ($process as $item) {
 							/** @var Coroutine\Socket $export */
 							$export = $item->exportSocket();
-							$export->send([$name => 'exit']);
+							$export->send(0x03455343213212);
 						}
 					}
 				}
@@ -100,6 +101,7 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess
 	 */
 	private function waiteExit(Process $process): void
 	{
+		$this->onProcessStop();
 		while ($this->isWorking()) {
 			$this->sleep();
 		}
diff --git a/System/Application.php b/System/Application.php
index 99a9a9f0..9ab13f6a 100644
--- a/System/Application.php
+++ b/System/Application.php
@@ -88,7 +88,7 @@ class Application extends BaseApplication
 	 */
 	public function addProcess(string $class, Process $process)
 	{
-		if (isset($this->_process[$class])) {
+		if (!isset($this->_process[$class])) {
 			$this->_process[$class] = [];
 		}
 		$this->_process[$class][] = $process;