Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Using redis streams to build more resilient ser...

Using redis streams to build more resilient services

Redis, das Schweizer Taschenmesser unter den Datenbanken, hat in der Version 5.0 einen neuen Datentyp dazu bekommen, der es wert ist, etwas genauer unter die Lupe genommen zu werden. Konkret geht es um Redis Streams und darum wie sie uns helfen, fehlertolerante und skalierbare Services zu bauen.

Thomas Schedler

September 27, 2019
Tweet

More Decks by Thomas Schedler

Other Decks in Programming

Transcript

  1. @chirimoya [email protected] https://github.com/chirimoya Hey, I’m Thomas – Co-founder & CEO

    of Sulu GmbH – More than 15 years of experience in web technologies & development – PHP, Symfony, React, SQL, Elasticsearch, … – Open source enthusiast – Loves cooking and mountains
  2. Tell, don’t ask. Generate an optimised read model for your

    service by consuming domain events from a central stream.
  3. “ In computer science, a stream is a sequence of

    data elements made available over time.
  4. Connect to Redis Assuming you have installed Redis server and

    client. And keep in mind, we need to use the latest version of Redis 5.0. PHP $redis = new \Redis(); $redis->connect('127.0.0.1', '6379'); CLI > redis-cli -h 127.0.0.1 -p 6379
  5. Add message to stream The stream elements can be one

    or more key-value pairs. Let's add elements to the stream: PHP $redis->xAdd('mystream', '*', [
 'KEY' => 'VALUE'
 ]); // MESSAGE ID (e.g. 1526919030474-55) $redis->xLen('mystream'); // (integer) 1 CLI > XADD mystream * KEY VALUE # MESSAGE ID (e.g. 1526919030474-55) > XLEN mystream # (integer) 1
  6. Read message from stream Read message from one or multiple

    streams, only returning entries with an ID greater than the last received ID reported by the caller. PHP $redis->xRead(['mystream' => 0], 2); // MESSAGES // [
 // 'STREAMNAME' => [
 // 'MESSAGEID' => [
 // 'KEY' => 'VALUE'
 // ],
 // 'MESSAGEID‘ => [
 // 'KEY' => 'VALUE'
 CLI > XREAD COUNT 2 STREAMS mystream 0 # MESSAGES # STREAMNAME [
 # MESSAGEID [
 # KEY
 # VALUE
 # ]
 # MESSAGEID [
 # KEY
  7. Block for data In order to avoid polling at a

    fixed or adaptive interval, the command is able to block until new entries get pushed into the requested stream. PHP $redis->xRead(['mystream' => '$'], 1, 0); // MESSAGES // [
 // 'STREAMNAME' => [
 // 'MESSAGEID' => [
 // 'KEY' => 'VALUE'
 // ]
 // ],
 // ] CLI > XREAD BLOCK 0 STREAMS mystream $ # MESSAGES # STREAMNAME [
 # MESSAGEID [
 # KEY
 # VALUE
 # ]
 # ]
  8. All clients are served with all the entries arriving in

    a stream. How can I scale my reads?
  9. Create a Consumer Group Redis stream provides the concept of

    consumer groups, allowing multiple consumers to process the same stream to implement load balancing. PHP $redis->xGroup(
 'CREATE', 
 'mystream',
 'mygroup'
 ); // 1/0 CLI > XGROUP CREATE mystream mygroup $ # 1/0
  10. Read data via Consumer Group Redis stream provides the concept

    of consumer groups, allowing multiple consumers to process the same stream to implement load balancing. PHP $redis->xReadGroup(
 'mygroup', 'myconsumer',
 ['mystream' => 0]
 ); // MESSAGES $redis->xReadGroup(
 'mygroup', 'myconsumer',
 ['mystream' => '>']
 ); CLI > XREADGROUP
 GROUP mygroup mycomsumer
 STREAMS mystream 0 # MESSAGES > XREADGROUP
 GROUP mygroup mycomsumer
 STREAMS mystream >
  11. Data processed confirmation Consumer groups require explicit acknowledgement of the

    messages successfully processed by the consumer. PHP $redis->xAck(
 'mystream', 
 'mygroup', 
 ['message-id']
 ); // 1/0 CLI > XACK mystream mygroup message-id # 1/0
  12. Failure Processing – Identify the message data that has been

    delivered but not confirmed – Change the owner of these data and redeliver PHP $redis->xPending('mystream', 'mygroup'); // MESSAGE IDs $redis->xClaim(
 'mystream', 
 'mygroup',
 'myconsumer',
 3600,
 ['message-id']
 ); CLI > XPENDING mystream mygroup # MESSAGE IDs > XCLAIM mystream mygroup myconsumer 3600 message-id
  13. Worked pretty good out of the box, but the API

    is quite low-level. Let's search for a high-level abstraction.
  14. <?php declare(strict_types=1); namespace HandcraftedInTheAlps\Bundle\RedisTransportBundle\Transport; use Symfony\Component\Messenger\Transport\TransportFactoryInterface; use Symfony\Component\Messenger\Transport\TransportInterface; class RedisStreamTransportFactory

    implements TransportFactoryInterface { public function createTransport(string $dsn, array $options): TransportInterface { $parsedUrl = parse_url($dsn); ... return new RedisStreamTransport($parsedUrl['host'], $parsedUrl['port'], ...); } public function supports(string $dsn, array $options): bool { return 0 === mb_strpos($dsn, 'redis-stream://'); } }
  15. # config/packages/messenger.yaml framework: messenger: buses: query_bus: ~ event_bus: ~ message_bus:

    middleware: - Avodi\Bundle\EventBundle\Common\DomainEvent\DomainEventMiddleware - App\Common\Messenger\SymfonyEventMiddleware - doctrine_transaction routing: 'Avodi\Bundle\EventBundle\Common\DomainEvent\DomainEventMessage': senders: ['domain_events'] transports: domain_events: 'redis://passwor@localhost:6379/mystream/mygroup/myconsumer?auto_setup=true'
  16. <?php class DomainEventCollector { /** * @var DomainEventMessage[] */ private

    $domainEvents = []; public function push(DomainEventMessage $event): void { $this->domainEvents[] = $event; } /** * @return DomainEventMessage[] */ public function release(): array { $domainEvents = $this->domainEvents; $this->domainEvents = []; return $domainEvents; } }
  17. public function cancelTicket(Ticket $ticket): void { $this->workflowRegistry->get($ticket)->apply($ticket, Ticket::TRANSITION_CANCEL); $newTicketId =

    Uuid::uuid4()->toString(); $this->createTicket($newTicketId, $ticket); $this->domainEventCollector->push( new DomainEventMessage( Ticket::class, 'canceled', $ticket->getId(), [ 'newTicketId' => $newTicketId, ] ) ); }
  18. <?php class TicketCanceledDomainEventMessageHandler extends AbstractTicketDomainEventMessageHandler { public function __invoke(DomainEventMessage $message):

    void { if ('canceled' !== $message->getType() || Ticket::class !== $message->getName()) { return; } $ticketIds = [$message->getId()]; $newTicketId = $message->getPayload()['newTicketId'] ?? ''; if ($newTicketId) { $ticketIds[] = $newTicketId; } else { $this->logger->error('Could not find newTicketId in canceled for access system.', [ 'canceledTicketId' => $message->getId(), ]); } $this->syncTickets($ticketIds); } }
  19. Trim Data from the stream When ~ argument is used,

    the trimming is performed only when Redis is able to remove a whole macro node. PHP $redis->xTrim('mystream', 1000, true); CLI > XTRIM mystream MAXLEN ~ 1000
  20. <?php class DomainEventHandler implements MessageHandlerInterface { /** * @var ElasticsearchStorage

    */ private $storage; public function __construct(ElasticsearchStorage $storage) { $this->storage = $storage; } public function __invoke(DomainEventMessage $message): void { $this->storage->store($message); } }