Merge branch 'master' into 3.2-merge

# Conflicts:
#	.github/workflows/test.yml
#	src/grpc-client/src/BaseClient.php

Author: limingxinleo

src/BaseClient.php

@@ -14,11 +14,13 @@
 
 use Google\Protobuf\Internal\Message;
 use Hyperf\Context\ApplicationContext;
+use Hyperf\Context\Context;
 use Hyperf\Coroutine\Channel\Pool as ChannelPool;
+use Hyperf\Coroutine\Locker;
 use Hyperf\Grpc\StatusCode;
 use Hyperf\GrpcClient\Exception\GrpcClientException;
 use InvalidArgumentException;
-use Swoole\Http2\Response;
+use Throwable;
 
 use function Hyperf\Support\retry;
 
@@ -29,17 +31,38 @@
  */
 class BaseClient
 {
-    private ?GrpcClient $grpcClient = null;
-
     private bool $initialized = false;
 
+    /**
+     * @var null|array<array-key,GrpcClient>
+     */
+    private ?array $grpcClients = null;
+
+    private int $clientCount = 1;
+
     public function __construct(private string $hostname, private array $options = [])
     {
+        $this->clientCount = max(1, (int) ($this->options['client_count'] ?? 0));
     }
 
     public function __destruct()
     {
-        $this->grpcClient?->close(false);
+        if (! $this->initialized) {
+            return;
+        }
+
+        $lastException = null;
+        foreach ($this->grpcClients as $client) {
+            try {
+                $client->close(false);
+            } catch (Throwable $exception) {
+                $lastException = $exception;
+            }
+        }
+
+        if ($lastException) {
+            throw $lastException;
+        }
     }
 
     public function __call(string $name, array $arguments): mixed
@@ -49,11 +72,13 @@ public function __call(string $name, array $arguments): mixed
 
     public function _getGrpcClient(): GrpcClient
     {
+        // Lazy initialization: defer client setup until first use to optimize resource usage.
         if (! $this->initialized) {
             $this->init();
         }
-        $this->start();
-        return $this->grpcClient;
+
+        // Ensure the client connection is started before use.
+        return $this->start();
     }
 
     /**
@@ -158,33 +183,65 @@ protected function _bidiRequest(
         return $call;
     }
 
-    private function start()
+    private function start(): GrpcClient
     {
-        $client = $this->grpcClient;
-        if (! ($client->isRunning() || $client->start())) {
-            $message = sprintf(
-                'Grpc client start failed with error code %d when connect to %s',
-                $client->getErrCode(),
-                $this->hostname
-            );
-            throw new GrpcClientException($message, StatusCode::INTERNAL);
+        $key = Context::getOrSet(self::class . '::id', fn () => array_rand($this->grpcClients));
+        $client = $this->grpcClients[$key];
+
+        // If the client is already running, return it directly.
+        if ($client->isRunning()) {
+            return $client;
         }
-        return true;
+
+        $lockKey = sprintf('%s:start:%d', spl_object_hash($this), $key);
+
+        if (Locker::lock($lockKey)) {
+            try {
+                $client->start(); // May throw exception
+            } catch (Throwable $e) {
+                $message = sprintf(
+                    'Grpc client start failed with error code %d when connect to %s',
+                    $client->getErrCode(),
+                    $this->hostname
+                );
+                throw new GrpcClientException($message, StatusCode::INTERNAL, $e);
+            } finally {
+                Locker::unlock($lockKey);
+            }
+        }
+
+        return $client;
     }
 
     private function init()
     {
-        if (! empty($this->options['client'])) {
-            if (! $this->options['client'] instanceof GrpcClient) {
-                throw new InvalidArgumentException('Parameter client have to instanceof Hyperf\GrpcClient\GrpcClient');
+        $lockKey = sprintf('%s:init', spl_object_hash($this));
+
+        if (Locker::lock($lockKey)) {
+            try {
+                if ($this->initialized) {
+                    return;
+                }
+
+                $channelPool = ApplicationContext::getContainer()->get(ChannelPool::class);
+                if (! empty($this->options['client'])) { // Use the specified client.
+                    if (! $this->options['client'] instanceof GrpcClient) {
+                        throw new InvalidArgumentException('Parameter client have to instanceof Hyperf\GrpcClient\GrpcClient');
+                    }
+                    $this->grpcClients[] = $this->options['client'];
+                } else { // Use multiple clients.
+                    for ($i = 0; $i < $this->clientCount; ++$i) {
+                        $grpcClient = new GrpcClient($channelPool);
+                        $grpcClient->set($this->hostname, $this->options);
+                        $this->grpcClients[] = $grpcClient;
+                    }
+                }
+
+                $this->initialized = true;
+            } finally {
+                Locker::unlock($lockKey);
             }
-            $this->grpcClient = $this->options['client'];
-        } else {
-            $this->grpcClient = new GrpcClient(ApplicationContext::getContainer()->get(ChannelPool::class));
-            $this->grpcClient->set($this->hostname, $this->options);
         }
-
-        $this->initialized = true;
     }
 
     private function buildRequest(string $method, Message $argument, array $options): Request

src/GrpcClient.php

@@ -97,11 +97,11 @@ public function set(string $hostname, array $options = [])
 
     public function start(): bool
     {
-        if ($this->recvCoroutineId !== 0 || $this->sendCoroutineId !== 0) {
-            throw new RuntimeException('Cannot restart the client.');
+        if ($this->recvCoroutineId !== 0 || $this->sendCoroutineId !== 0) { // Client already started
+            return true;
         }
         if (! Coroutine::inCoroutine()) {
-            throw new RuntimeException('Client must be started in coroutine');
+            throw new RuntimeException('Client must be started in coroutine.');
         }
         if (! $this->getHttpClient()->connect()) {
             throw new GrpcClientException('Connect failed, error=' . $this->getHttpClient()->errMsg, $this->getHttpClient()->errCode);
@@ -149,7 +149,7 @@ public function closeRecv()
 
     public function isConnected(): bool
     {
-        return $this->httpClient->connected;
+        return (bool) $this->httpClient?->connected;
     }
 
     public function isStreamExist(int $streamId): bool