as2252258/kiri-rpc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

qqq

Browse Source
This commit is contained in:
向林
2023-07-31 23:09:00 +08:00
parent 493a58281f
commit 10a64ee646
7 changed files with 287 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Consul.php
-103
View File
@@ -1,103 +0,0 @@
<?php
namespace Kiri\Rpc;
use Exception;
use Kiri;
use Kiri\Abstracts\Component;
use Kiri\Consul\Agent;
use Psr\Container\ContainerExceptionInterface;
use Kiri\Di\ContainerInterface;
use Psr\Container\NotFoundExceptionInterface;
use Swoole\Coroutine;
class Consul extends Component
{
public Agent $agent;
private array $config = [];
/**
* @param ContainerInterface $container
* @param array $settings
* @param array $config
* @throws Exception
*/
public function __construct(public ContainerInterface $container, array $settings, array $config = [])
{
parent::__construct($config);
$this->config = $settings;
}
/**
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
public function init(): void
{
$this->agent = $this->container->get(Agent::class);
}
/**
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
public function deregister(): void
{
if (env('environmental') != Kiri::WORKER && env('environmental_workerId') != 0) {
return;
}
$agent = $this->container->get(Agent::class);
$this->logger->debug("disconnect consul.");
$agent->service->deregister($this->config['ID']);
$agent->checks->deregister($this->config['Check']['CheckId']);
}
/**
* @return void
*/
public function service_health(): void
{
$info = $this->agent->service->service_health($this->config['ID']);
if ($info->getStatusCode() == 200) {
return;
}
$this->agent->service->register($this->config);
}
public function watches()
{
}
/**
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
public function register(): void
{
$this->deregister();
$data = $this->agent->service->register($this->config);
if ($data->getStatusCode() != 200) {
$this->logger->error($data->getBody());
}
}
}
LotteryDraw.php
+185
View File
@@ -0,0 +1,185 @@
<?php
namespace Kiri\Rpc;
class LotteryDraw
{
/**
* @param array $goods
* @param string $wight
* @return mixed
* @uses $reward = Lucked::luck($data);
*/
public static function luck(array $goods, string $wight = 'probability'): mixed
{
static $class = null;
if ($class === null) $class = new LotteryDraw();
if (count($goods) < 1) return null;
$prob = $alias = [];
$class->ket($array = $class->init($goods, $wight), $prob, $alias);
$result = $class->generation($array, $prob, $alias);
if (!isset($goods[$result])) {
return null;
}
return $goods[$result];
}
/**
* @param array $goods
* @param string $wight
* @return array
*/
protected function init(array $goods, string $wight): array
{
$defaultIndex = -1;
$array = [];
foreach ($goods as $key => $val) {
if ($val[$wight] == 0) $defaultIndex = $key;
$array[] = (float)$val[$wight];
}
$total = array_sum($array);
if ($total > 1) {
$array = $this->reset($array, $total);
$total = array_sum($array);
}
if (1 - $total !== floatval(0)) {
if ($defaultIndex == -1) {
$defaultIndex = count($goods);
}
$array[$defaultIndex] = 1 - array_sum($array);
}
return $array;
}
/**
* @param array $data
* @param float $total
* @return array
*/
protected function reset(array $data, float $total): array
{
$length = $this->getMin($data);
foreach ($data as $key => $val) {
$data[$key] = ($val * $length - ($val * $length) / ($total * $length)) / $length;
}
return $data;
}
/**
* @param $data
* @param $prob
* @param $alias
* @return mixed
*/
private function generation($data, $prob, $alias): mixed
{
$nums = count($prob) - 1;
$MAX_P = $this->getMin($data); // 假设最小的几率是万分之一
$coin_toss = mt_rand(1, $MAX_P) / $MAX_P; // 抛出硬币
$col = mt_rand(0, $nums); // 随机落在一列
$b_head = $coin_toss > $prob[$col]; // 判断是否落在原色
return $b_head ? $alias[$col] : $col;
}
/**
* @param $num
* @return int
*/
private function getMin($num): int
{
$length = $this->getFloatLength($this->mini($num));
return (int)sprintf('1%0' . $length . 'd', 0);
}
/**
* @param $num
* @return int|float
*/
protected function mini($num): int|float
{
$def = current($num);
foreach ($num as $val) {
if ($val < $def) {
$def = $val;
}
}
return $def;
}
/**
* @param $float
* @return int
*/
private function getFloatLength($float): int
{
$ex = explode('.', $float);
return strlen(end($ex));
}
/**
* @param array $data
* @param array $prob
* @param array $alias
*/
private function ket(array $data, array &$prob, array &$alias): void
{
$nums = count($data);
$small = $large = [];
for ($i = 0; $i < $nums; ++$i) {
$data[$i] = $data[$i] * $nums; // 扩大倍数，使每列高度可为1
/** 分到两个数组，便于组合 */
if ($data[$i] < 1) {
$small[] = $i;
} else {
$large[] = $i;
}
}
/** 将超过1的色块与原色拼凑成1 */
while (!empty($small) && !empty($large)) {
$n_index = array_shift($small);
$a_index = array_shift($large);
$prob[$n_index] = $data[$n_index];
$alias[$n_index] = $a_index;
// 重新调整大色块
$data[$a_index] = ($data[$a_index] + $data[$n_index]) - 1;
if ($data[$a_index] < 1) {
$small[] = $a_index;
} else {
$large[] = $a_index;
}
}
/** 剩下大色块都设为1 */
while (!empty($large)) {
$n_index = array_shift($large);
$prob[$n_index] = 1;
}
/** 一般是精度问题才会执行这一步 */
while (!empty($small)) {
$n_index = array_shift($small);
$prob[$n_index] = 1;
}
}
}
Luckdraw.php
-151
View File
@@ -1,151 +0,0 @@
<?php
namespace Kiri\Rpc;
use Exception;
class Luckdraw
{
/**
* @param array $goods
* @param string $wight
* @return mixed
* @array = [
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ['name'=> '商品名称', 'probability'=> '概率', 'total'=> '库存'],
* ]
*
* @uses $reward = Lucked::luck($data);
*/
public static function luck(array $goods, string $wight = 'probability'): mixed
{
static $class = null;
if ($class === null) $class = new Luckdraw();
if (empty($goods)) return null;
$array = $prob = $alias = [];
$defaultIndex = 0;
foreach ($goods as $key => $val) {
if ($val[$wight] == 0) $defaultIndex = $key;
$array[] = (float)$val[$wight];
}
$array[$defaultIndex] = 1 - array_sum($array);
$class->ket($array, $prob, $alias);
$result = $class->generation($array, $prob, $alias);
if (!isset($goods[$result])) {
return null;
}
return $goods[$result];
}
/**
* @param $data
* @param $prob
* @param $alias
* @return mixed
*/
private function generation($data, $prob, $alias): mixed
{
$nums = count($prob) - 1;
$MAX_P = $this->getMin($data); // 假设最小的几率是万分之一
$coin_toss = rand(1, $MAX_P) / $MAX_P; // 抛出硬币
$col = rand(0, $nums); // 随机落在一列
$b_head = $coin_toss < $prob[$col]; // 判断是否落在原色
return $b_head ? $col : @$alias[$col];
}
/**
* @param $num
* @return string
*/
private function getMin($num): string
{
$def = current($num);
foreach ($num as $val) {
if ($val < $def) {
$def = $val;
}
}
$length = $this->getFloatLength($def) + 1;
return sprintf('1%0' . $length . 'd', 0);
}
/**
* @param $float
* @return int
*/
private function getFloatLength($float): int
{
$ex = explode('.', 1 - $float);
return strlen(end($ex));
}
/**
* @param array $data
* @param array $prob
* @param array $alias
*/
private function ket(array $data, array &$prob, array &$alias)
{
$nums = count($data);
$small = $large = [];
for ($i = 0; $i < $nums; ++$i) {
$data[$i] = $data[$i] * $nums; // 扩大倍数，使每列高度可为1
/** 分到两个数组，便于组合 */
if ($data[$i] < 1) {
$small[] = $i;
} else {
$large[] = $i;
}
}
/** 将超过1的色块与原色拼凑成1 */
while (!empty($small) && !empty($large)) {
$n_index = array_shift($small);
$a_index = array_shift($large);
$prob[$n_index] = $data[$n_index];
$alias[$n_index] = $a_index;
// 重新调整大色块
$data[$a_index] = ($data[$a_index] + $data[$n_index]) - 1;
if ($data[$a_index] < 1) {
$small[] = $a_index;
} else {
$large[] = $a_index;
}
}
/** 剩下大色块都设为1 */
while (!empty($large)) {
$n_index = array_shift($large);
$prob[$n_index] = 1;
}
/** 一般是精度问题才会执行这一步 */
while (!empty($small)) {
$n_index = array_shift($small);
$prob[$n_index] = 1;
}
}
}
RpcJsonp.php
+24 -54
View File
@@ -7,26 +7,23 @@ use JetBrains\PhpStorm\ArrayShape;
use Kiri;
use Kiri\Di\LocalService;
use Kiri\Abstracts\Component;
use Kiri\Annotation\Annotation;
use Kiri\Consul\Agent;
use Kiri\Core\Json;
use Kiri\Events\EventProvider;
use Kiri\Exception\ConfigException;
use Kiri\Message\Handler\DataGrip;
use Kiri\Message\Handler\Router;
use Kiri\Message\Handler\RouterCollector;
use Kiri\Server\Contract\OnCloseInterface;
use Kiri\Server\Contract\OnConnectInterface;
use Kiri\Server\Contract\OnReceiveInterface;
use Kiri\Server\Events\OnBeforeShutdown;
use Kiri\Server\Events\OnServerBeforeStart;
use Psr\Container\ContainerExceptionInterface;
use Kiri\Di\ContainerInterface;
use Psr\Container\NotFoundExceptionInterface;
use ReflectionException;
use Swoole\Coroutine;
use Kiri\Router\RouterCollector;
use Psr\Container\ContainerInterface;
use Swoole\Coroutine\Channel;
use Swoole\Server;
use Kiri\Router\DataGrip;
/**
*
@@ -35,46 +32,36 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
{
private array $consul = [];
public RouterCollector $collector;
/**
* @param ContainerInterface $container
* @param Router $router
* @param Annotation $annotation
* @param DataGrip $dataGrip
* @param RpcManager $manager
* @param EventProvider $eventProvider
* @param array $config
* @throws Exception
*/
public function __construct(public ContainerInterface $container,
public Router $router,
public Annotation $annotation,
public DataGrip $dataGrip,
public RpcManager $manager,
public EventProvider $eventProvider,
array $config = [])
public EventProvider $eventProvider)
{
parent::__construct($config);
parent::__construct();
}
/**
* @return void
* @throws ConfigException
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
/**
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
* @throws ReflectionException
* @throws Exception
*/
public function init(): void
{
$this->eventProvider->on(OnBeforeShutdown::class, [$this, 'onBeforeShutdown']);
// $this->consul = \config('rpc.consul', null);
// if (!empty($this->consul)) {
// $this->eventProvider->on(OnServerBeforeStart::class, [$this, 'register']);
// }
$this->collector = $this->dataGrip->get('rpc');
$this->registerConsumers();
}
@@ -82,8 +69,7 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
/**
* @return void
* @throws ConfigException
* @throws ContainerExceptionInterface
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
public function registerConsumers(): void
@@ -100,40 +86,24 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
}
/**
* @param OnBeforeShutdown $beforeShutdown
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
*/
/**
* @param OnBeforeShutdown $beforeShutdown
* @return void
*/
public function onBeforeShutdown(OnBeforeShutdown $beforeShutdown): void
{
if (env('environmental') != Kiri::WORKER && env('environmental_workerId') != 0) {
return;
}
$agent = $this->container->get(Agent::class);
$this->logger->debug("disconnect consul.");
$agent->service->deregister($this->consul['ID']);
$agent->checks->deregister($this->consul['Check']['CheckId']);
}
/**
* @param OnServerBeforeStart $server
* @throws ConfigException
*/
public function register(OnServerBeforeStart $server)
{
$consumers = \config("rpc.consumers", []);
if (!empty($consumers)) {
foreach ($consumers as $service => $consumer) {
$this->manager->add($service, $consumer);
}
}
$this->manager->register($this->consul);
}
@@ -168,7 +138,6 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
}
return $server->send($fd, $this->batchDispatch($data), $reactor_id);
} catch (\Throwable $throwable) {
$this->logger->error('JsonRpc: ' . $throwable->getMessage());
$response = Json::encode($this->failure(-32700, $throwable->getMessage()));
return $server->send($fd, $response, $reactor_id);
}
@@ -291,11 +260,12 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
}
/**
* @param int $fd
* @return void
*/
public function OnClose(int $fd): void
/**
* @param Server $server
* @param int $fd
* @return void
*/
public function OnClose(Server $server, int $fd): void
{
// TODO: Implement onClose() method.
}
RpcManager.php
+24
View File
@@ -2,8 +2,10 @@
namespace Kiri\Rpc;
use Etcd\Client;
use Kiri;
use Kiri\Abstracts\Component;
use PhpParser\Node\Stmt\Return_;
/**
@@ -12,4 +14,26 @@ use Kiri\Abstracts\Component;
class RpcManager extends Component
{
protected array $services = [];
/**
* @return void
*/
public function watch(): void
{
$data = new Client('','');
}
/**
* @param $service
* @return mixed
*/
public function getServices($service): mixed
{
return $this->services[$service] ?? null;
}
}
RpcProcess.php
+53
View File
@@ -0,0 +1,53 @@
<?php
namespace Kiri\Rpc;
use Kiri\Di\Context;
use Kiri\Server\Abstracts\BaseProcess;
use Swoole\Coroutine;
use Swoole\Process;
class RpcProcess extends BaseProcess
{
/**
* @return string
*/
public function getName(): string
{
return "Rpc Manager";
}
/**
* @param Process|null $process
* @return void
*/
public function process(?Process $process): void
{
// TODO: Implement process() method.
while (true) {
$read = $process->read();
}
}
/**
* @return $this
*/
public function onSigterm(): static
{
// TODO: Implement onSigterm() method.
if (Context::inCoroutine()) {
Coroutine::create(fn() => $this->onShutdown(Coroutine::waitSignal(SIGTERM | SIGINT)));
} else {
pcntl_signal(SIGTERM, [$this, 'onStop']);
pcntl_signal(SIGINT, [$this, 'onStop']);
}
return $this;
}
}
TraitTransporter.php
+1 -1
View File
@@ -72,7 +72,7 @@ trait TraitTransporter
if (count($array) < 2) {
$luck = $array[0];
} else {
$luck = Luckdraw::luck($array, 'Weight');
$luck = LotteryDraw::luck($array, 'Weight');
}
return [
'Address' => $luck['TaggedAddresses']['wan_ipv4']['Address'],