This commit is contained in:
2021-04-12 11:05:57 +08:00
parent 996ff50e99
commit c62be31e6e
4 changed files with 43 additions and 38 deletions
+25 -24
View File
@@ -7,6 +7,7 @@ namespace HttpServer\Events;
use Exception;
use HttpServer\Abstracts\Callback;
use Kafka\Struct;
use Snowflake\Crontab\Crontab;
use Snowflake\Event;
use Snowflake\Exception\ComponentException;
use Swoole\Server;
@@ -22,16 +23,15 @@ class OnPipeMessage extends Callback
* @param Server $server
* @param int $src_worker_id
* @param $message
* @throws ComponentException
* @throws Exception
*/
public function onHandler(Server $server, int $src_worker_id, $message)
{
try {
$swoole_unserialize = swoole_unserialize($message);
match ($swoole_unserialize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swoole_unserialize),
'crontab' => $this->onCrontabWorker($swoole_unserialize),
$swollen_universalize = swoole_unserialize($message);
match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $message)
};
} catch (\Throwable $exception) {
@@ -42,26 +42,28 @@ class OnPipeMessage extends Callback
}
/**
* @param array $message
* @return string
*/
private function onCrontabWorker(array $message)
/**
* @param array $message
* @return string
* @throws Exception
*/
private function onCrontabWorker(array $message): string
{
/** @var \Snowflake\Crontab\Crontab $crontab */
/** @var Crontab $crontab */
$crontab = $message['handler'] ?? null;
$crontab->increment()->execute();
return 'success';
}
/**
* @param $server
* @param $src_worker_id
* @param $message
* @throws \Exception
*/
private function onMessageWorker($server, $src_worker_id, $message)
/**
* @param $server
* @param $src_worker_id
* @param $message
* @return string
* @throws Exception
*/
private function onMessageWorker($server, $src_worker_id, $message): string
{
fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]);
@@ -69,12 +71,11 @@ class OnPipeMessage extends Callback
}
/**
* @param array $message
* @throws \ReflectionException
* @throws \Snowflake\Exception\NotFindClassException
*/
private function onKafkaWorker(array $message)
/**
* @param array $message
* @return string
*/
private function onKafkaWorker(array $message): string
{
[$topic, $rdMessage] = $message['body'];
+5 -4
View File
@@ -6,6 +6,7 @@ namespace Snowflake\Crontab;
use Closure;
use Exception;
use JetBrains\PhpStorm\Pure;
use Snowflake\Abstracts\BaseObject;
use Snowflake\Event;
use Snowflake\Snowflake;
@@ -64,9 +65,9 @@ class Crontab extends BaseObject
/**
* @return string
*/
public function getName(): string
#[Pure] public function getName(): string
{
return $this->name;
return md5($this->name);
}
/**
@@ -203,7 +204,7 @@ class Crontab extends BaseObject
{
$redis = Snowflake::app()->getRedis();
$redis->set('crontab:' . ($name = md5($this->getName())), swoole_serialize($this));
$redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this));
$tickTime = time() + $this->getTickTime();
@@ -219,7 +220,7 @@ class Crontab extends BaseObject
try {
$redis = Snowflake::app()->getRedis();
$name_md5 = md5($this->getName());
$name_md5 = $this->getName();
$redis->hSet(self::WAIT_END, $name_md5, serialize($this));
$params = call_user_func($this->handler, $this->params, $this->name);
+1 -1
View File
@@ -27,7 +27,7 @@ class Producer extends Component
{
$redis = Snowflake::app()->getRedis();
$name = md5($crontab->getName());
$name = $crontab->getName();
$redis->set('crontab:' . $name, swoole_serialize($crontab));
+12 -9
View File
@@ -10,6 +10,7 @@ use Snowflake\Snowflake;
use Swoole\Coroutine\WaitGroup;
use Swoole\Coroutine\Channel;
use Swoole\Timer;
use Snowflake\Cache\Redis;
/**
* Class Zookeeper
@@ -43,7 +44,7 @@ class Zookeeper extends Process
name($this->pid, 'Crontab zookeeper.');
}
Timer::tick(1000, function () {
[$range, $redis] = $this->loadCrotabTask();
[$range, $redis] = $this->loadCarobTask();
$server = Snowflake::app()->getSwoole();
$setting = $server->setting['worker_num'];
@@ -55,12 +56,14 @@ class Zookeeper extends Process
}
/**
* @param $server
* @param $redis
* @throws \Exception
*/
private function dispatch($server, $redis, $setting, $value)
/**
* @param $server
* @param Redis|\Redis $redis
* @param int $setting
* @param $value
* @throws Exception
*/
private function dispatch($server, Redis|\Redis $redis, int $setting, $value)
{
$server->sendMessage(swoole_serialize([
'action' => 'crontab', 'handler' => swoole_unserialize($redis->get('crontab:' . $value))
@@ -71,9 +74,9 @@ class Zookeeper extends Process
/**
* @return array
* @throws \Exception
* @throws Exception
*/
private function loadCrotabTask()
private function loadCarobTask(): array
{
$redis = Snowflake::app()->getRedis();