Redis Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require enqueue/redis
   

   Requires php-redis extension (ext-redis) and Redis server (v2.6+).

2. Basic Configuration:

   use Enqueue\Redis\RedisConnectionFactory;
   use Enqueue\Redis\RedisContext;
   
   $redis = new Redis();
   $redis->connect('127.0.0.1', 6379);
   
   $connectionFactory = new RedisConnectionFactory($redis);
   $connection = $connectionFactory->createConnection();
   

3. First Use Case:

   // Producer: Send a message
   $context = new RedisContext();
   $producer = $connection->createProducer();
   $producer->send(new Message('Hello Redis Queue!'));
   
   // Consumer: Process messages
   $consumer = $connection->createConsumer('queue_name');
   $message = $consumer->receive();
   if ($message) {
       echo $message->getBody();
       $consumer->ack($message);
   }
   

Where to Look First

• Documentation for API reference and examples.
• Queue Interop Spec for standard message handling.
• src/ directory for core classes (RedisConnectionFactory, RedisContext, RedisConnection).

Implementation Patterns

Core Workflows

1. Producer Pattern:

   $producer = $connection->createProducer();
   $producer->send(new Message('data', ['priority' => 5])); // Optional headers
   

   • Use send() for fire-and-forget.
   • Use sendWithDelay() for delayed messages (Redis 2.6+):

     $producer->sendWithDelay(new Message('delayed'), 10); // 10 seconds
     

2. Consumer Pattern:

   $consumer = $connection->createConsumer('queue_name');
   while ($message = $consumer->receive()) {
       try {
           // Process message
           $consumer->ack($message);
       } catch (\Exception $e) {
           $consumer->nack($message); // Reject with requeue
       }
   }
   

   • Ack/Nack: Always handle messages in a try-catch to avoid lost messages.
   • Batch Processing: Use receiveNoWait() for non-blocking checks.

3. Connection Management:

   $connection = $connectionFactory->createConnection(['queue' => 'my_queue']);
   $connection->close(); // Explicitly close when done
   

   • Reuse connections for performance (Redis connections are pooled).

Integration with Laravel

1. Service Provider:

   use Enqueue\Redis\RedisConnectionFactory;
   use Illuminate\Support\ServiceProvider;
   
   class QueueServiceProvider extends ServiceProvider {
       public function register() {
           $redis = new \Redis();
           $redis->connect(config('redis.default.host'));
           $this->app->singleton('redis.connection', function () {
               return (new RedisConnectionFactory($redis))->createConnection();
           });
       }
   }
   

2. Laravel Queue Integration:

   // In config/queue.php
   'connections' => [
       'redis' => [
           'driver' => 'enqueue',
           'connection' => 'redis.connection',
       ],
   ],
   

   • Use php artisan queue:work redis to run workers.

3. Delayed Jobs:

   $job = (new MyJob())->delay(now()->addMinutes(5));
   Queue::push('redis', $job);
   

Advanced Patterns

• Fanout Exchanges: Use RedisConnection::createProducer('exchange_name') for pub/sub.
• Dead Letter Queues (DLQ):

  $consumer = $connection->createConsumer('queue_name', ['deadLetterQueue' => 'dlq']);
  

• Message TTL:

  $producer->send(new Message('data', [], 3600)); // Expires in 1 hour
  

Gotchas and Tips

Common Pitfalls

1. Connection Issues:

   • Symptom: RedisException: Connection refused.
   • Fix: Ensure Redis server is running and accessible. Use ping() to test:

     $redis->ping(); // Should return true
     

   • Timeouts: Set connection timeout in RedisConnectionFactory:

     $connectionFactory = new RedisConnectionFactory($redis, [
         'read_write_timeout' => 5.0,
     ]);
     

2. Message Loss:

   • Cause: Unhandled exceptions in consumers (no ack/nack).
   • Fix: Always wrap message processing in try-catch and explicitly ack/nack.

3. Priority Queues:

   • Redis does not natively support priority queues. Use sorted sets (e.g., RedisPriorityQueue) or external libraries like php-enqueue/priority-queue.

4. Memory Leaks:

   • Cause: Large messages or unclosed connections.
   • Fix: Limit message size (e.g., serialize payloads) and explicitly close connections:

     $connection->close();
     

5. Serialization:

   • Issue: Redis stores messages as raw strings. Use Message::create() with custom serializers:

     $message = Message::create('data', [], new JsonSerializer());
     

Debugging Tips

1. Inspect Queues:

   redis-cli LRANGE queue_name 0 -1  # List all messages
   redis-cli HGETALL message:uuid   # Inspect message metadata
   

2. Enable Logging:

   $connection = $connectionFactory->createConnection([
       'logger' => new \Monolog\Logger('redis', [new \Monolog\Handler\StreamHandler('php://stderr')]),
   ]);
   

3. Check for Blocked Consumers:

   • Use redis-cli MONITOR to observe consumer activity.
   • Reset stuck consumers with:

     redis-cli BRPOPLPUSH queue_name temp_queue 0
     

Configuration Quirks

1. Redis Cluster Support:

   • Use predis/predis with RedisConnectionFactory:

     $client = new \Predis\Client(['scheme' => 'tcp', 'host' => '127.0.0.1']);
     $connectionFactory = new RedisConnectionFactory($client);
     

2. SSL/TLS:

   • Configure Redis client for secure connections:

     $redis = new \Redis();
     $redis->connect('redis.example.com', 6379, null, null, \Redis::OPT_SSL);
     

3. Environment-Specific Config:

   • Use Laravel's .env for Redis settings:

     REDIS_HOST=127.0.0.1
     REDIS_PORT=6379
     REDIS_PASSWORD=secret
     

   • Load in RedisConnectionFactory:

     $redis = new \Redis();
     $redis->connect(env('REDIS_HOST'), env('REDIS_PORT'), null, null, \Redis::OPT_AUTH, env('REDIS_PASSWORD'));
     

Extension Points

1. Custom Context:

   • Extend RedisContext to add middleware:

     class CustomRedisContext extends RedisContext {
         public function __construct() {
             $this->middleware([]); // Add middleware
         }
     }
     

2. Message Serialization:

   • Implement MessageSerializerInterface for custom formats:

     class ProtobufSerializer implements MessageSerializerInterface {
         public function decode($data) { /* ... */ }
         public function encode($message) { /* ... */ }
     }
     

   • Attach to producer/consumer:

     $producer = $connection->createProducer(['serializer' => new ProtobufSerializer()]);
     

3. Event Listeners:

   • Subscribe to queue events (e.g., message sent/received):

     $connection->getEventDispatcher()->addListener('message.sent', function ($event) {
         // Log or track sent messages
     });