ForumPHP 2023 : Nous Sous-Estimons Tous Redis

Redis, ce n'est pas que du cache. C'est évidemment beaucoup de cache. C'est d'ailleurs son utilisation principale (voire unique) dans beaucoup d'applications.

Mais connaissez-vous les streams ? Ces structures de données un peu particulières intégrées à Redis prouvent que ce dernier peut être utilisé pour bien plus que ça. Écriture de logs, aggrégation de données, notifications, buffer temporaire pour l'écriture d'un énorme fichier, les cas d'usages sont infinis. On peut facilement imaginer qu'en couplant les Streams aux générateurs de PHP, les performances d'un tel système peuvent dépasser toutes les attentes.

Depuis quelques années, la bibliothèque PHP Predis prend en charge les Redis Streams et permet leur utilisation avec beaucoup d'aisance. Il est temps de donner un second souffle à nos scripts PHP utilisant intensément Redis... ou qui devraient l'utiliser !

Transcript

1. REDIS. NOUS SOUS- ESTIMONS TOUS ALEXANDRE DAUBOIS FORUMPHP 2023
2. SYMFONY LEAD DEVELOPER CONTRIBUTEUR À PHP ET SYMFONY Alexandre Daubois @ALEXDAUBOIS
5. JSON. Des énormes fichiers
6. DEMANDE DE DÉPLOIEMENT GÉNÉRATION DU JSON EN MÉMOIRE DANS UN UNIQUE PROCESS UPLOAD VERS LE CLOUD
7. GÉNÉRATION DU JSON EN MÉMOIRE DANS UN UNIQUE PROCESS Fatal error: Allowed memory size of [beaucoup] bytes exhausted (tried to allocate [plus que beaucoup] bytes)
8. ON VA ÉCRIRE DANS UN FICHIER TEMPORAIRE !
9. Aujourd'hui Demain Le problème
10. Aujourd'hui Demain Le problème
11. LE CLOUD. Puis un jour,
13. Aujourd'hui Demain Le problème
14. Aujourd'hui Demain Le problème (urgent)
15. OÙ ÉCRIRE NOTRE FICHIER EN CONSTRUCTION ? CLOUD = ∞ PODS, COMMENT PARALLÉLISER ?
16. SYSTÈME DE FICHIER ? BEAUCOUP TROP LENT (+ PAS ADAPTÉ AU CLOUD)
17. EN MÉMOIRE ? RAPIDE MAIS LIMITÉ PAR PHP (ET TOUJOURS PAS ADAPTÉ AU CLOUD)
18. EN BASE DE DONNÉES ? PARTAGÉ MAIS TROP INTENSE POUR L'I/O
19. EN MÉMOIRE PARTAGÉ GRANDE CAPACITÉ
20. REDIS. BONJOUR
22. CLÉ VALEUR
23. $ redis-cli redis 127.0.0.1:6379> ping PONG redis 127.0.0.1:6379> set mykey somevalue OK redis 127.0.0.1:6379> get mykey "somevalue"
24. Streams
25. STRUCTURE DE DONNÉES APPEND-ONLY IDENTIFIANT PAR ORDRE CHRONOLOGIQUE D'AJOUT RAPIDE. RAPIDE. RAPIDE.
26. 1 MILLION D'INSERTIONS PAR SECONDE. (SUR UNE MACHINE "MOYENNE")
27. ... DEMANDE DE DÉPLOIEMENT CONCATÉNATION DU STREAM
28. ... DEMANDE DE DÉPLOIEMENT CONCATÉNATION DU STREAM
29. #[AsMessageHandler] readonly class SerializeComponentMessageHandler { // ... public function __invoke(SerializeComponentMessage $message): void { try { // Conversion en JSON $serialized = $this->serializer->serialize($message->getComponent()); // Ajout au stream $streamId = 'catalog-deployment-'.$message->getId(); $this->redis->xadd($streamId, ['data' => $serialized]); // Vérification de consolidation if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); } } catch (\Exception $exception) { throw new UnrecoverableMessageHandlingException($exception); } } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
30. #[AsMessageHandler] readonly class SerializeComponentMessageHandler { // ... public function __invoke(SerializeComponentMessage $message): void { try { // Conversion en JSON $serialized = $this->serializer->serialize($message->getComponent()); // Ajout au stream $streamId = 'catalog-deployment-'.$message->getId(); $this->redis->xadd($streamId, ['data' => $serialized]); // Vérification de consolidation if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); } } catch (\Exception $exception) { throw new UnrecoverableMessageHandlingException($exception); } } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 // Ajout au stream $streamId = 'catalog-deployment-'.$message->getId(); $this->redis->xadd($streamId, ['data' => $serialized]); #[AsMessageHandler] 1 readonly class SerializeComponentMessageHandler 2 { 3 // ... 4 5 public function __invoke(SerializeComponentMessage $message): void 6 { 7 try { 8 // Conversion en JSON 9 $serialized = $this->serializer->serialize($message->getComponent()); 10 11 12 13 14 15 // Vérification de consolidation 16 if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { 17 $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); 18 } 19 } catch (\Exception $exception) { 20 throw new UnrecoverableMessageHandlingException($exception); 21 } 22 } 23 } 24
31. #[AsMessageHandler] readonly class SerializeComponentMessageHandler { // ... public function __invoke(SerializeComponentMessage $message): void { try { // Conversion en JSON $serialized = $this->serializer->serialize($message->getComponent()); // Ajout au stream $streamId = 'catalog-deployment-'.$message->getId(); $this->redis->xadd($streamId, ['data' => $serialized]); // Vérification de consolidation if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); } } catch (\Exception $exception) { throw new UnrecoverableMessageHandlingException($exception); } } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 // Ajout au stream $streamId = 'catalog-deployment-'.$message->getId(); $this->redis->xadd($streamId, ['data' => $serialized]); #[AsMessageHandler] 1 readonly class SerializeComponentMessageHandler 2 { 3 // ... 4 5 public function __invoke(SerializeComponentMessage $message): void 6 { 7 try { 8 // Conversion en JSON 9 $serialized = $this->serializer->serialize($message->getComponent()); 10 11 12 13 14 15 // Vérification de consolidation 16 if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { 17 $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); 18 } 19 } catch (\Exception $exception) { 20 throw new UnrecoverableMessageHandlingException($exception); 21 } 22 } 23 } 24 // Vérification de consolidation if ($this->consolidationChecker->shouldConsolidate(/** ... */)) { $this->messageBus->dispatch(new ConsolidateCatalogMessage(/** ... */); } #[AsMessageHandler] 1 readonly class SerializeComponentMessageHandler 2 { 3 // ... 4 5 public function __invoke(SerializeComponentMessage $message): void 6 { 7 try { 8 // Conversion en JSON 9 $serialized = $this->serializer->serialize($message->getComponent()); 10 11 // Ajout au stream 12 $streamId = 'catalog-deployment-'.$message->getId(); 13 $this->redis->xadd($streamId, ['data' => $serialized]); 14 15 16 17 18 19 } catch (\Exception $exception) { 20 throw new UnrecoverableMessageHandlingException($exception); 21 } 22 } 23 } 24
32. On concatène le stream...
33. Fatal error: Allowed memory size of [beaucoup] bytes exhausted (tried to allocate [plus que beaucoup] bytes)
34. ÉTAPE. UNE DERNIÈRE
35. ÉCRITURE PAR CHUNKS LECTURE D'UN BOUT (CHUNK) DE STREAM ÉCRITURE DANS LE CLOUD
36. GÉNÉRATEUR, OU "FONCTION GÉNÉRATRICE" foreach ($generator as $v) EXÉCUTE JUSQU'AU PROCHAIN "yield" VALEUR RETOURNÉE + EXÉCUTION DU CORPS DE LA BOUCLE
37. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
38. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29
39. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29
40. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29
41. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29
42. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 $this->redis->xdel($streamId, $redisId); } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 16 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29
43. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 $this->redis->xdel($streamId, $redisId); } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 16 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29
44. final class JsonRedisStreamEncoder implements StreamEncoderInterface { public function __construct(private ClientInterface $redis) {} public function encode(string $streamId): \Generator { $index = 0; yield '['; while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; $this->redis->xdel($streamId, $redisId); } $this->redis->del($streamId); yield ']'; } } $encoder = new JsonRedisStreamEncoder(); foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public function encode(string $streamId): \Generator { $index = 0; yield '['; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 5 6 7 8 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29 $this->redis->xdel($streamId, $redisId); } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 16 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { $redisId = \array_key_first($component); yield (++$index > 1 ? ',' : '').$data; final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 11 12 13 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 foreach ($encoder->encode('stream-id') as $chunk) { 27 // Upload $chunk 28 } 29 foreach ($encoder->encode('stream-id') as $chunk) { // Upload $chunk } final class JsonRedisStreamEncoder implements StreamEncoderInterface 1 { 2 public function __construct(private ClientInterface $redis) {} 3 4 public function encode(string $streamId): \Generator 5 { 6 $index = 0; 7 8 yield '['; 9 10 while ([] !== ($component = $this->redis->xrange($streamId, '0', '+', 1))) { 11 $redisId = \array_key_first($component); 12 13 yield (++$index > 1 ? ',' : '').$data; 14 15 $this->redis->xdel($streamId, $redisId); 16 } 17 18 $this->redis->del($streamId); 19 20 yield ']'; 21 } 22 } 23 24 $encoder = new JsonRedisStreamEncoder(); 25 26 27 28 29
45. Be lazy, use STREAM. LAZY
46. use LazyStream\Exception\LazyStreamWriterTriggerException; use LazyStream\LazyStreamWriter; // ... $dataProvider = (new JsonRedisStreamEncoder($redisClient))->encode('my-stream-id'); // Déclaration d'un lazy stream... $lazyStream = new LazyStreamWriter('gs://my-bucket/catalog.json', $dataProvider); try { $lazyStream->trigger(); } catch (LazyStreamWriterTriggerException) { // ... } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 STREAM. LAZY
47. use LazyStream\Exception\LazyStreamWriterTriggerException; use LazyStream\MultiLazyStreamWriter; // ... $dataProvider = (new JsonRedisStreamEncoder($redisClient))->encode('my-stream-id'); // Possibilité de multiplexing si besoin $lazyStream = new MultiLazyStreamWriter([ 'gs://.../catalog.json', 's3://.../catalog.json', ], $dataProvider); try { $lazyStream->trigger(); } catch (LazyStreamWriterTriggerException) { // ... } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 STREAM. LAZY
48. composer require alexandre-daubois/lazy-stream Be lazy, use STREAM. LAZY https://github.com/alexandre-daubois/lazy-stream
49. REDIS. MERCI
50. EN MÉMOIRE PARTAGÉ GRANDE CAPACITÉ
51. UN FEEDBACK ?
52. MERCI @ALEXDAUBOIS FORUMPHP 2023