Redis

2025-07-18

25 min

Redis Stream Veri Türü

Redis Streams ile real-time data processing, message passing ve event sourcing. XADD, XREAD komutları ile stream-based architectures.

Redis

Streams

Real-time

Processing

Events

Redis'in sunduğu en gelişmiş append-only log veri yapılarından biri olan Stream, time-series events ve real-time data processing için tasarlanmış sofistike bir veri türüdür. Stream, chronological ordering, unique message IDs, consumer groups ve persistent message queues özellikleri ile event sourcing, message streaming, real-time analytics ve distributed processing senaryolarında vazgeçilmez bir araçtır. Her message otomatik olarak timestamp-based unique ID alır ve stream içinde immutable olarak saklanır. Bu özelliği sayesinde event replay, audit trails, real-time monitoring ve microservices communication gibi modern architecture patterns'larını native olarak destekler.

📋 Redis Stream Veri Türü Özellikleri Tablosu

Özellik	Açıklama	Örnek
Append-only structure	Immutable, sadece ekleme yapılabilen log yapısı	Events sonradan değiştirilemez
Unique message IDs	Her message otomatik timestamp-sequence ID alır	`1642678800000-0` (timestamp-sequence)
Consumer groups	Distributed message processing ve load balancing	`XGROUP CREATE stream group $ MKSTREAM`
Range queries	Zaman bazlı message range'leri okuyabilme	`XRANGE events 1640995200000 1640998800000`
Blocking operations	Real-time event listening ve streaming	`XREAD BLOCK 0 STREAMS events $`
Message acknowledgment	Delivered message tracking ve redelivery	`XACK group consumer message-id`
Persistent storage	Message'lar disk'te persist edilir	Restart sonrası messages korunur
Memory optimization	Configurable trimming ve automatic cleanup	`XTRIM events MAXLEN ~ 1000`

🎯 Redis Stream Kullanım Senaryoları

Kullanım Alanı	Açıklama	Redis Komutu Örneği
Event sourcing	Domain events'lerin sequential saklanması	`XADD events * event_type "user_created"`
Message queues	Asynchronous task processing	`XADD tasks * task_type "send_email"`
Real-time analytics	Live data streaming ve processing	`XADD metrics * cpu_usage "75.5"`
IoT data collection	Sensor verilerinin continuous collection	`XADD sensors * temperature "23.5"`
Audit trails	System activity logging ve compliance	`XADD audit * action "user_login"`
Chat applications	Real-time messaging systems	`XADD chat:room123 * message "Hello"`
Financial transactions	Transaction log ve reconciliation	`XADD transactions * amount "100.50"`
Game events	Player actions ve game state changes	`XADD game:events * action "level_up"`
Log aggregation	Centralized logging from multiple services	`XADD logs * service "api" level "error"`
CDC (Change Data Capture)	Database change tracking	`XADD changes * table "users" op "UPDATE"`

Stream'in en güçlü özelliklerinden biri, consumer groups mekanizmasıdır. Bu özellik sayesinde multiple consumers arasında load balancing, message acknowledgment ve redelivery işlemlerini otomatik olarak yönetebilirsiniz. Consumer group'lar, distributed processing, fault tolerance ve horizontal scaling için kritik öneme sahiptir. Bir consumer crash olsa bile, acknowledge edilmemiş message'lar otomatik olarak diğer consumer'lara redirect edilir.
Stream'in bir diğer devrimsel özelliği ise time-based range queries'dir. XRANGE ve XREVRANGE komutları ile belirli zaman aralıklarındaki message'ları millisecond precision ile çekebilirsiniz. Bu özellik, time-series analytics, historical data analysis ve event replay senaryoları için muazzam avantaj sağlar. Örneğin, "son 1 saatteki tüm error eventleri" veya "belirli user'ın bugünkü tüm aktiviteleri" gibi sorguları nanosecond seviyesinde hızla çözebilirsiniz.
🧩 1. Stream Veri Türü Temelleri
Redis'teki Stream, time-ordered, append-only log yapısıdır. Her entry (message) bir unique ID ve field-value çiftlerinden oluşur. Message ID'ler otomatik olarak timestamp-sequence formatında generate edilir ve chronological ordering garanti edilir. Stream'ler immutable'dır, yani eklenen message'lar sonradan değiştirilemez.
🔧 Temel Özellikler
Append-only: Sadece ekleme yapılabilir, modification mümkün değil
Unique IDs: Her message benzersiz timestamp-sequence ID alır
Time-ordered: Message'lar kronolojik sırada tutulur
Persistent: Message'lar disk'te kalıcı olarak saklanır
Field-value structure: Her message birden fazla field içerebilir

🔧 Temel Stream Komutları ve Açıklamaları

Komut	Açıklama	Örnek
`XADD`	Stream'e yeni message ekler	`XADD events * type "login"`
`XREAD`	Stream'den message'ları okur	`XREAD STREAMS events 0`
`XRANGE`	Belirli ID range'indeki message'ları getirir	`XRANGE events - +`
`XREVRANGE`	Reverse chronological order'da message'lar	`XREVRANGE events + - COUNT 10`
`XLEN`	Stream'deki message sayısını verir	`XLEN events`
`XGROUP`	Consumer group'ları yönetir	`XGROUP CREATE events group1 $`
`XREADGROUP`	Consumer group ile message okur	`XREADGROUP GROUP group1 c1`
`XACK`	Message'ı acknowledge eder	`XACK events group1 msg-id`
`XPENDING`	Acknowledge bekleyen message'ları listeler	`XPENDING events group1`
`XTRIM`	Stream'i belirli boyuta kırpar	`XTRIM events MAXLEN 1000`
`XINFO`	Stream hakkında bilgi verir	`XINFO STREAM events`
`XDEL`	Message'ları siler	`XDEL events message-id`

🧪 XADD Komutunu İnceleyelim

Stream'e yeni message eklemek için kullanılır.

text
127.0.0.1:6379> XADD events * event_type "user_login" user_id "123" timestamp "2024-01-20T10:30:00Z"  // Events stream'ine login eventi ekliyoruz.
"1642678200000-0"                                         // Otomatik generate edilen unique message ID.
127.0.0.1:6379> XADD events * event_type "page_view" user_id "456" page "/homepage"  // Page view eventi ekliyoruz.
"1642678201000-0"                                         // Yeni message ID (timestamp arttı).
127.0.0.1:6379> XADD events 1642678202000-0 event_type "purchase" user_id "123" amount "99.99"  // Manuel ID ile message ekliyoruz.
"1642678202000-0"                                         // Belirtilen ID kullanıldı.
127.0.0.1:6379> XADD metrics * cpu_usage "75.5" memory_usage "60.2" disk_usage "45.8"  // Metrics stream'ine sistem bilgileri.
"1642678203000-0"                                         // Metrics için yeni message ID.
127.0.0.1:6379> XADD chat:room1 * user "alice" message "Hello everyone!" timestamp "2024-01-20T10:35:00Z"  // Chat message'ı.
"1642678500000-0"                                         // Chat stream'inde message ID.

📌 Dikkat Edilmesi Gereken Noktalar
Otomatik ID generation * kullandığınızda Redis otomatik olarak current timestamp ile unique ID generate eder.

Manuel ID specification

text
127.0.0.1:6379>
127.0.0.1:6379> XADD events 1000000000000-0 type "test"   // Manuel olarak ID belirtiyoruz.
127.0.0.1:6379>

ID format: timestamp-sequence Format: millisecond_timestamp-sequence_number

Field-value çiftleri Her message birden fazla field-value çifti içerebilir:

text
127.0.0.1:6379>
127.0.0.1:6379> XADD orders * customer_id "C123" product "laptop" quantity "2" total "2999.98"
127.0.0.1:6379>

Stream yoksa otomatik oluşturulur İlk XADD komutu ile stream otomatik olarak oluşturulur.

🧪 XREAD Komutunu İnceleyelim

Stream'den message'ları okumak için kullanılır.

bash
127.0.0.1:6379> XREAD STREAMS events 0                   // events stream'ini baştan okuyoruz.
1) 1) "events"                                            // Stream adı.
   2) 1) 1) "1642678200000-0"                             // Message ID.
         2) 1) "event_type"                               // Field 1.
            2) "user_login"                               // Value 1.
            3) "user_id"                                  // Field 2.
            4) "123"                                      // Value 2.
            5) "timestamp"                                // Field 3.
            6) "2024-01-20T10:30:00Z"                     // Value 3.
      2) 1) "1642678201000-0"                             // İkinci message ID.
         2) 1) "event_type"
            2) "page_view"
            3) "user_id"
            4) "456"
            5) "page"
            6) "/homepage"
127.0.0.1:6379> XREAD COUNT 1 STREAMS events 0           // Sadece 1 message okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678200000-0"
         2) 1) "event_type"
            2) "user_login"
            3) "user_id"
            4) "123"
            5) "timestamp"
            6) "2024-01-20T10:30:00Z"
127.0.0.1:6379> XREAD BLOCK 5000 STREAMS events $        // 5 saniye boyunca yeni message'ları bekle.
(nil)                                                     // 5 saniyede yeni message gelmedi.
127.0.0.1:6379> XREAD STREAMS events metrics 0 0         // Birden fazla stream'i aynı anda okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678200000-0"
         2) 1) "event_type"
            2) "user_login"
            ... (events stream messages)
2) 1) "metrics"
   2) 1) 1) "1642678203000-0"
         2) 1) "cpu_usage"
            2) "75.5"
            ... (metrics stream messages)

📌 Dikkat Edilmesi Gereken Noktalar
Starting ID parametresi 0 = en baştan oku, $ = en son message'dan sonra oku (yeni message'ları bekle).

BLOCK seçeneği ile real-time reading

bash
127.0.0.1:6379>
127.0.0.1:6379> XREAD BLOCK 0 STREAMS events $           // Sonsuza kadar yeni message bekle.
127.0.0.1:6379>

COUNT ile limit setting

text
127.0.0.1:6379>
127.0.0.1:6379> XREAD COUNT 5 STREAMS events 0           // Maksimum 5 message oku.
127.0.0.1:6379>

Multiple stream reading Birden fazla stream'i aynı anda monitor edebilirsiniz.

🧪 XRANGE ve XREVRANGE Komutlarını İnceleyelim

Belirli ID aralığındaki message'ları okur.

text
127.0.0.1:6379> XRANGE events - +                        // Tüm message'ları kronolojik sırada getir.
1) 1) "1642678200000-0"
   2) 1) "event_type"
      2) "user_login"
      3) "user_id"
      4) "123"
      5) "timestamp"
      6) "2024-01-20T10:30:00Z"
2) 1) "1642678201000-0"
   2) 1) "event_type"
      2) "page_view"
      3) "user_id"
      4) "456"
      5) "page"
      6) "/homepage"
127.0.0.1:6379> XRANGE events 1642678200000 1642678201000  // Belirli timestamp aralığı.
1) 1) "1642678200000-0"
   2) 1) "event_type"
      2) "user_login"
      ... (message details)
2) 1) "1642678201000-0"
   2) 1) "event_type"
      2) "page_view"
      ... (message details)
127.0.0.1:6379> XRANGE events - + COUNT 2                // İlk 2 message'ı getir.
1) 1) "1642678200000-0"
   ... (first message)
2) 1) "1642678201000-0"
   ... (second message)
127.0.0.1:6379> XREVRANGE events + - COUNT 1             // En son message'ı getir (reverse order).
1) 1) "1642678202000-0"
   2) 1) "event_type"
      2) "purchase"
      3) "user_id"
      4) "123"
      5) "amount"
      6) "99.99"

📌 Dikkat Edilmesi Gereken Noktalar
Special boundaries - = stream'in başlangıcı, + = stream'in sonu.

Timestamp-based queries

text
127.0.0.1:6379>
127.0.0.1:6379> XRANGE events (1642678200000 1642678202000  // Exclusive range (1642678200000 hariç).
127.0.0.1:6379>

COUNT limitation Performance için COUNT kullanarak result set'i sınırlayın.
XREVRANGE reverse order En yeni message'lardan geriye doğru okuma için kullanın.

🧪 Consumer Groups - XGROUP ve XREADGROUP

Distributed message processing için consumer group'lar oluşturur.

bash
127.0.0.1:6379> XGROUP CREATE events processors $ MKSTREAM  // "processors" consumer group'u oluşturuyoruz.
OK                                                        // Group başarıyla oluşturuldu.
127.0.0.1:6379> XADD events * task_type "send_email" recipient "user@example.com" priority "high"  // Yeni task ekliyoruz.
"1642678300000-0"
127.0.0.1:6379> XREADGROUP GROUP processors worker1 COUNT 1 STREAMS events >  // worker1 olarak message okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678300000-0"
         2) 1) "task_type"
            2) "send_email"
            3) "recipient"
            4) "user@example.com"
            5) "priority"
            6) "high"
127.0.0.1:6379> XADD events * task_type "generate_report" user_id "789" format "pdf"  // Başka bir task.
"1642678301000-0"
127.0.0.1:6379> XREADGROUP GROUP processors worker2 COUNT 1 STREAMS events >  // worker2 olarak okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678301000-0"
         2) 1) "task_type"
            2) "generate_report"
            3) "user_id"
            4) "789"
            5) "format"
            6) "pdf"
127.0.0.1:6379> XACK events processors 1642678300000-0   // worker1'in task'ını acknowledge ediyoruz.
(integer) 1                                               // 1 message acknowledge edildi.
127.0.0.1:6379> XPENDING events processors               // Pending (acknowledge edilmemiş) message'ları kontrol ediyoruz.
1) (integer) 1                                            // 1 pending message var.
2) "1642678301000-0"                                     // En eski pending message ID.
3) "1642678301000-0"                                     // En yeni pending message ID.
4) 1) 1) "worker2"                                        // Pending message'ı alan consumer.
      2) "1"                                             // Bu consumer'ın pending message sayısı.

📌 Dikkat Edilmesi Gereken Noktalar
Consumer group creation $ ile group'u stream'in en sonundan başlatıyoruz (sadece yeni message'ları okur).
Multiple workers Aynı group'taki farklı consumer'lar farklı message'ları alır (load balancing).
Message acknowledgment XACK ile message'ın işlendiğini belirtmek zorundasınız.
Pending messages Acknowledge edilmemiş message'lar XPENDING ile takip edilir.

Consumer failure handling

text
127.0.0.1:6379>
127.0.0.1:6379> XCLAIM events processors worker1 300000 1642678301000-0  // 5 dakika sonra message'ı başka worker'a transfer et.
127.0.0.1:6379>

🧪 XLEN ve XTRIM Komutlarını İnceleyelim

Stream bilgileri ve memory management için kullanılır.

text
127.0.0.1:6379> XLEN events                              // events stream'indeki message sayısı.
(integer) 4                                               // 4 message var.
127.0.0.1:6379> XADD events * test_type "load_test"      // Load test için çok sayıda message ekliyoruz.
127.0.0.1:6379> XADD events * test_type "load_test"
127.0.0.1:6379> XADD events * test_type "load_test"
127.0.0.1:6379> XLEN events                              // Güncel message sayısı.
(integer) 7
127.0.0.1:6379> XTRIM events MAXLEN 5                    // Stream'i 5 message ile sınırla.
(integer) 2                                               // 2 eski message silindi.
127.0.0.1:6379> XLEN events                              // Trimming sonrası message sayısı.
(integer) 5
127.0.0.1:6379> XTRIM events MAXLEN ~ 3                  // Approximately 3 message kalsın (~: yaklaşık).
(integer) 2                                               // 2 message daha silindi.
127.0.0.1:6379> XLEN events
(integer) 3

📌 Dikkat Edilmesi Gereken Noktalar
MAXLEN ile exact trimming Tam olarak belirtilen sayıda message kalır.
~ ile approximate trimming Performance için yaklaşık trimming yapar (daha hızlı).

Automatic trimming

text
127.0.0.1:6379>
127.0.0.1:6379> XADD events MAXLEN ~ 1000 * type "auto_trim"  // Add sırasında otomatik trim.
127.0.0.1:6379>

Memory optimization Büyük stream'ler için düzenli trimming yapın.
🔄 Stream ile Gerçek Dünya Senaryoları

📊 Senaryo 1: E-commerce Order Processing Pipeline

bash
127.0.0.1:6379> XGROUP CREATE orders order_processors $ MKSTREAM  // Order processing group'u oluşturuyoruz.
OK
127.0.0.1:6379> XADD orders * order_id "ORD001" customer_id "C123" total "299.99" status "pending"  // Yeni order.
"1642678400000-0"
127.0.0.1:6379> XADD orders * order_id "ORD002" customer_id "C456" total "149.50" status "pending"
"1642678401000-0"
127.0.0.1:6379> XREADGROUP GROUP order_processors payment_service COUNT 1 STREAMS orders >  // Payment service order alıyor.
1) 1) "orders"
   2) 1) 1) "1642678400000-0"
         2) 1) "order_id"
            2) "ORD001"
            3) "customer_id"
            4) "C123"
            5) "total"
            6) "299.99"
            7) "status"
            8) "pending"
# Payment service order'ı işledikten sonra:
127.0.0.1:6379> XACK orders order_processors 1642678400000-0  // Order acknowledge ediliyor.
(integer) 1
127.0.0.1:6379> XADD orders * order_id "ORD001" customer_id "C123" total "299.99" status "paid"  // Payment completed eventi.
"1642678402000-0"
127.0.0.1:6379> XREADGROUP GROUP order_processors inventory_service COUNT 1 STREAMS orders >  // Inventory service sıradaki order'ı alıyor.
1) 1) "orders"
   2) 1) 1) "1642678401000-0"
         2) 1) "order_id"
            2) "ORD002"
            ... (order details)
127.0.0.1:6379> XINFO GROUPS orders                      // Consumer group bilgilerini kontrol ediyoruz.
1) 1) "name"
   2) "order_processors"
   3) "consumers"
   4) (integer) 2                                         // 2 aktif consumer.
   5) "pending"
   6) (integer) 1                                         // 1 pending message.

🔥 Senaryo 2: Real-time IoT Sensor Data Collection

bash
127.0.0.1:6379> XGROUP CREATE sensor_data data_processors $ MKSTREAM  // Sensor data processing group.
OK
127.0.0.1:6379> XADD sensor_data * sensor_id "TEMP001" location "warehouse_a" temperature "23.5" humidity "65.2"  // Sıcaklık sensörü.
"1642678500000-0"
127.0.0.1:6379> XADD sensor_data * sensor_id "PRES001" location "tank_1" pressure "145.7" unit "psi"  // Basınç sensörü.
"1642678501000-0"
127.0.0.1:6379> XADD sensor_data * sensor_id "TEMP001" location "warehouse_a" temperature "24.1" humidity "66.8"  // Temp update.
"1642678502000-0"
# Real-time monitoring:
127.0.0.1:6379> XREAD BLOCK 1000 STREAMS sensor_data $   // 1 saniye boyunca yeni sensor data bekle.
# Analytics service specific sensor'ları okur:
127.0.0.1:6379> XREADGROUP GROUP data_processors analytics_worker COUNT 2 STREAMS sensor_data >
1) 1) "sensor_data"
   2) 1) 1) "1642678500000-0"
         2) 1) "sensor_id"
            2) "TEMP001"
            ... (sensor data)
      2) 1) "1642678501000-0"
         2) 1) "sensor_id"
            2) "PRES001"
            ... (pressure data)
# Time-range analytics:
127.0.0.1:6379> XRANGE sensor_data 1642678500000 1642678502000  // Son 2 saniyedeki tüm sensor readings.
1) 1) "1642678500000-0"
   ... (temp data 1)
2) 1) "1642678501000-0"
   ... (pressure data)
3) 1) "1642678502000-0"
   ... (temp data 2)
127.0.0.1:6379> XACK sensor_data data_processors 1642678500000-0 1642678501000-0  // Batch acknowledge.
(integer) 2

💬 Senaryo 3: Real-time Chat Application

markdown
127.0.0.1:6379> XADD chat:room1 * user_id "alice" username "Alice" message "Hello everyone!" timestamp "2024-01-20T11:00:00Z"  // Chat message.
"1642680000000-0"
127.0.0.1:6379> XADD chat:room1 * user_id "bob" username "Bob" message "Hi Alice!" timestamp "2024-01-20T11:00:15Z"
"1642680015000-0"
127.0.0.1:6379> XADD chat:room1 * user_id "charlie" username "Charlie" message "Good morning!" timestamp "2024-01-20T11:00:30Z"
"1642680030000-0"
# Real-time message listening:
127.0.0.1:6379> XREAD BLOCK 0 STREAMS chat:room1 1642680030000-0  // Son message'dan sonraki yeni message'ları bekle.
# Message history for new user:
127.0.0.1:6379> XREVRANGE chat:room1 + - COUNT 10        // Son 10 message'ı getir (history).
1) 1) "1642680030000-0"
   2) 1) "user_id"
      2) "charlie"
      3) "username"
      4) "Charlie"
      5) "message"
      6) "Good morning!"
      7) "timestamp"
      8) "2024-01-20T11:00:30Z"
2) 1) "1642680015000-0"
   ... (Bob's message)
3) 1) "1642680000000-0"
   ... (Alice's message)
# Message count:
127.0.0.1:6379> XLEN chat:room1                          // Toplam message sayısı.
(integer) 3
# Auto-cleanup (retain last 100 messages):
127.0.0.1:6379> XTRIM chat:room1 MAXLEN ~ 100            // Son 100 message'ı tut.
(integer) 0                                               // Silinecek message yok (henüz 3 tane var).

🏦 Senaryo 4: Financial Transaction Audit Trail

bash
127.0.0.1:6379> XGROUP CREATE transactions audit_processors $ MKSTREAM  // Audit processing group.
OK
127.0.0.1:6379> XADD transactions * transaction_id "TXN001" account_from "ACC123" account_to "ACC456" amount "1000.00" type "transfer"  // Transfer transaction.
"1642680100000-0"
127.0.0.1:6379> XADD transactions * transaction_id "TXN002" account "ACC789" amount "500.00" type "deposit" source "external"
"1642680101000-0"
127.0.0.1:6379> XADD transactions * transaction_id "TXN003" account "ACC123" amount "150.00" type "withdrawal" location "ATM_001"
"1642680102000-0"
# Fraud detection service:
127.0.0.1:6379> XREADGROUP GROUP audit_processors fraud_detector COUNT 5 STREAMS transactions >
1) 1) "transactions"
   2) 1) 1) "1642680100000-0"
         2) 1) "transaction_id"
            2) "TXN001"
            ... (transaction details)
      2) 1) "1642680101000-0"
         ... (deposit transaction)
      3) 1) "1642680102000-0"
         ... (withdrawal transaction)
# Account-specific transaction history:
127.0.0.1:6379> XRANGE transactions - +                  // Tüm transactions (audit için).
# Compliance reporting (daily transactions):
127.0.0.1:6379> XRANGE transactions 1642636800000 1642723199999  // Belirli gün aralığındaki transactions.
127.0.0.1:6379> XACK transactions audit_processors 1642680100000-0 1642680101000-0 1642680102000-0  // Audit complete.
(integer) 3
127.0.0.1:6379> XINFO STREAM transactions                // Transaction stream bilgileri.
1) "length"
2) (integer) 3
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1642680102000-0"
9) "groups"
10) (integer) 1                                          // 1 consumer group.

Redis Stream veri türü, event-driven architectures ve real-time data processing konusunda Redis'in en gelişmiş ve güçlü araçlarından biridir. Append-only log yapısı, consumer groups ve time-based queries ile modern application patterns'larını native olarak destekler. Bu bölümde XADD, XREAD, XRANGE, XGROUP, XREADGROUP gibi temel komutları ve gerçek dünya senaryolarında Stream kullanımını detaylı örneklerle öğrendik. Stream'in append-only structure, consumer groups, time-based queries ve message acknowledgment özellikleri özellikle:
- Event sourcing ve CQRS patterns
- Message queues ve async processing
- Real-time analytics ve monitoring
- IoT data collection ve processing
- Chat applications ve real-time communication
- Financial transaction logging
- Audit trails ve compliance gibi senaryolarda vazgeçilmez bir araçtır.
Stream veri türü, event-driven applications ihtiyaçlarınız için Redis'te en comprehensive çözümü sunar. Time-ordered persistence, distributed processing ve real-time capabilities üçlüsü, modern applications'ların complex event handling requirements'larını karşılamak için mükemmel bir kombinasyon oluşturur. Bu bölümde öğrendiğiniz komutlar ve senaryolar ile artık:
Event sourcing ve CQRS pattern implementasyonları geliştirebilir
Message queue sistemleri ve async processing pipelines tasarlayabilir
Real-time data streaming applications oluşturabilir
IoT data collection ve sensor monitoring sistemleri geliştirebilir
Audit trail ve transaction logging mekanizmaları implementasyonları yapabilirsiniz Stream'leri projelerinizde etkin kullanarak, kullanıcılarınıza event-driven, real-time ve highly-scalable deneyimler sunabilirsiniz. Sırada ne var? Bir sonraki yazımızda Redis'in Geo veri türünü derinlemesine inceleyeceğiz. Geo, geospatial indexing ve location-based queries için tasarlanmış specialized bir veri türüdür. GPS coordinates'leri nasıl verimli şekilde saklayabileceğimizi, proximity searches nasıl yapabileceğimizi, radius-based queries ile location services nasıl geliştirebileceğimizi ve geospatial applications'ın mobile era'daki kritik önemini öğreneceğiz.
Konuya başlamadan önce çok fazla özel isim kullandığımı fark ettim. Bu konuda elimden geldiğince sizlere tek tek bu kavramları öncelikle açıklamak istiyorum.
📝 Ek Bilgi: Önemli Kavramlar
📚 Temel Yapı Kavramları 📝 Append-only Log
Append-only = Sadece ekleme yapılabilir, silme/değiştirme olmaz
Log = Kronolojik sırayla tutulan kayıt defteri
Örnek: Banka hesap hareketleri gibi → Her işlem eklenir, değiştirilmez
Avantaj: Veri bütünlüğü ve audit trail garantisi
🔗 Chronological Ordering
Chronological = Zaman sırasına göre dizilmiş
Ordering = Sıralama sistemi
Redis'te her mesaj zaman damgası alır ve kronolojik sırada tutulur
Böylece "10:30'daki mesajlar" veya "bugünkü tüm olaylar" sorgulanabilir
🆔 Unique Message IDs
Her mesaj benzersiz ID alır: 1642678200000-0 formatında
İlk kısım: Timestamp (Unix millisecond)
İkinci kısım: Sequence number (aynı millisecond'deki sıra)
Hiçbir mesaj aynı ID'ye sahip olamaz
🏗️ Field-value Çiftleri
Her mesaj anahtar-değer çiftlerinden oluşur
Örnek: user_id: "123", action: "login", time: "10:30"
JSON'a benzer ama daha hızlı ve Redis'e optimize
🔒 Immutable
Immutable = Değiştirilemez, sabit
Mesaj eklendikten sonra içeriği değiştirilemez
Sadece XDEL ile silinebilir
Veri güvenliği ve tutarlılık için kritik
⏱️ Timestamp-sequence
Timestamp: Unix zaman damgası (millisecond)
Sequence: Aynı anda gelen mesajları ayırt etmek için
Format: <timestamp>-<sequence>
Otomatik generate edilir veya manuel verilebilir
🔄 İşlemci/Consumer Kavramları
👥 Consumer Groups
Consumer = Mesaj işleyen uygulama/servis
Group = Birlikte çalışan consumer'lar topluluğu
Her mesaj sadece grup içindeki 1 consumer tarafından işlenir
Load balancing ve parallel processing sağlar
⚖️ Load Balancing
Load = İş yükü/veri miktarı
Balancing = Dengeleme/dağıtma
Mesajlar consumer'lar arasında eşit dağıtılır
1000 mesaj + 5 consumer = Her biri ~200 mesaj alır
🔄 Redelivery
Redelivery = Yeniden teslim etme
Consumer crash olursa, işlenmemiş mesajlar başka consumer'a verilir
Veri kaybı olmaz, güvenilirlik artar
Otomatik failover mekanizması
🌐 Distributed Processing
Distributed = Dağıtık/çoklu makine
Processing = İşleme/hesaplama
Mesajlar farklı sunuculardaki consumer'lar tarafından işlenir
Horizontal scaling ve performance artışı
📈 Horizontal Scaling
Horizontal = Yatay (makine sayısı artırma)
Scaling = Ölçeklendirme/büyütme
Daha fazla consumer ekleyerek kapasiteyi artırma
Vertical scaling'in tersi (tek makinede RAM/CPU artırma)
🛡️ Fault Tolerance
Fault tolerance (hata toleransı), sistemin bir veya birden fazla bileşeni arızalandığında bile çalışmaya devam etme yeteneğidir
Redis Stream'lerde, bir consumer crash olsa bile diğer consumer'lar işi devralabilir
Örnek: E-ticaret siparişi işleyen 3 consumer var → 1'i çökerse → kalan 2'si işleme devam eder
Bu sayede sistem 7/24 kesintisiz çalışır ve veri kaybı olmaz
⏰ Zaman Tabanlı Kavramlar
🔍 Time-based Range Queries
Time-based = Zaman temelli
Range = Aralık/başlangıç-bitiş
Queries = Sorgular
Belirli zaman aralığındaki mesajları getirme
Örnek: "14:00-15:00 arası tüm error'lar"
📊 Time-series Analytics
Time-series = Zaman serisi verileri
Analytics = Analiz/çözümleme
Zamana göre değişen verilerin analizi
Örnek: CPU kullanımı, satış rakamları, sensör verileri
📚 Historical Data Analysis
Historical = Geçmiş/tarihsel
Data Analysis = Veri analizi
Geçmiş verileri inceleyip pattern bulma
Örnek: "Geçen ay en çok hangi saatlerde login olmuş?"
🔄 Event Replay
Event = Olay/durum değişikliği
Replay = Yeniden oynatma
Geçmiş olayları tekrar işleme/simülasyon
Debugging ve testing için kritik
✅ Mesaj Durumu Kavramları
✅ Message Acknowledgment
Message acknowledgment (mesaj onaylama), bir consumer'ın mesajı başarıyla işlediğini bildirmesidir
Consumer mesajı aldıktan sonra XACK komutu ile "işledim" der
Eğer consumer crash olur ve acknowledge etmezse, mesaj pending (beklemede) kalır
Pending mesajlar otomatik olarak diğer consumer'lara redelivery (yeniden teslim) edilir
⏳ Pending
Pending = Beklemede/askıda
Consumer'a verilmiş ama henüz acknowledge edilmemiş mesajlar
Consumer crash olursa bu mesajlar diğer consumer'lara verilir
XPENDING komutu ile görülebilir
💾 Kalıcılık Kavramları
💿 Persistent Storage
Persistent = Kalıcı/sürekli
Storage = Depolama
Mesajlar RAM'de değil disk'te saklanır
Server restart olsa bile mesajlar korunur
🗂️ Persistent Message Queues
Message Queues = Mesaj kuyrukları
Disk'te kalıcı olarak saklanan mesaj kuyrukları
Reliability ve durability garantisi
Elektrik kesilse bile mesajlar kaybolmaz
🏃‍♂️ Real-time Kavramları
⚡ Real-time Analytics
Real-time = Gerçek zamanlı/anlık
Analytics = Analiz/çözümleme
Veriler gelir gelmez anında işleme
Millisecond seviyesinde hızlı yanıt
📞 Blocking Operations
Blocking = Bekletme/engelleyici
Operations = İşlemler
Yeni mesaj gelene kadar bekleme yapma
XREAD BLOCK komutu ile gerçek zamanlı dinleme
🎯 Uygulama Alanları
📝 Event Sourcing
Event Sourcing = Olay kaynağı
Sistemdeki her değişikliği event olarak saklama
State yerine events tutma
Audit, replay ve debugging avantajları
📨 Message Queues
Message = Mesaj/veri paketi
Queues = Kuyruklar
Asynchronous (eş zamanlı olmayan) işlem kuyruğu
Producer-Consumer pattern
🏭 IoT Data Collection
IoT = Internet of Things (Nesnelerin İnterneti)
Data Collection = Veri toplama
Sensör verilerinin sürekli toplanması
Temperature, humidity, motion verileri
🧾 Audit Trails
Audit = Denetim/kontrol
Trails = İzler/kayıtlar
Sistem üzerindeki tüm işlemlerin kaydı
Compliance ve security için kritik
💰 Financial Transactions
Financial = Finansal/parasal
Transactions = İşlemler
Para transferi, ödeme kayıtları
ACID properties ve integrity garantisi
🔄 CDC (Change Data Capture)
Change Data Capture = Veri değişikliği yakalama
Database'deki değişiklikleri anlık yakalama
INSERT, UPDATE, DELETE operasyonları
Data synchronization için kullanılır

Redis

2025-07-18

25 min

Redis Stream Veri Türü

Redis Streams ile real-time data processing, message passing ve event sourcing. XADD, XREAD komutları ile stream-based architectures.

Redis

Streams

Real-time

Processing

Events

Redis'in sunduğu en gelişmiş append-only log veri yapılarından biri olan Stream, time-series events ve real-time data processing için tasarlanmış sofistike bir veri türüdür. Stream, chronological ordering, unique message IDs, consumer groups ve persistent message queues özellikleri ile event sourcing, message streaming, real-time analytics ve distributed processing senaryolarında vazgeçilmez bir araçtır. Her message otomatik olarak timestamp-based unique ID alır ve stream içinde immutable olarak saklanır. Bu özelliği sayesinde event replay, audit trails, real-time monitoring ve microservices communication gibi modern architecture patterns'larını native olarak destekler.

📋 Redis Stream Veri Türü Özellikleri Tablosu

Özellik	Açıklama	Örnek
Append-only structure	Immutable, sadece ekleme yapılabilen log yapısı	Events sonradan değiştirilemez
Unique message IDs	Her message otomatik timestamp-sequence ID alır	`1642678800000-0` (timestamp-sequence)
Consumer groups	Distributed message processing ve load balancing	`XGROUP CREATE stream group $ MKSTREAM`
Range queries	Zaman bazlı message range'leri okuyabilme	`XRANGE events 1640995200000 1640998800000`
Blocking operations	Real-time event listening ve streaming	`XREAD BLOCK 0 STREAMS events $`
Message acknowledgment	Delivered message tracking ve redelivery	`XACK group consumer message-id`
Persistent storage	Message'lar disk'te persist edilir	Restart sonrası messages korunur
Memory optimization	Configurable trimming ve automatic cleanup	`XTRIM events MAXLEN ~ 1000`

🎯 Redis Stream Kullanım Senaryoları

Kullanım Alanı	Açıklama	Redis Komutu Örneği
Event sourcing	Domain events'lerin sequential saklanması	`XADD events * event_type "user_created"`
Message queues	Asynchronous task processing	`XADD tasks * task_type "send_email"`
Real-time analytics	Live data streaming ve processing	`XADD metrics * cpu_usage "75.5"`
IoT data collection	Sensor verilerinin continuous collection	`XADD sensors * temperature "23.5"`
Audit trails	System activity logging ve compliance	`XADD audit * action "user_login"`
Chat applications	Real-time messaging systems	`XADD chat:room123 * message "Hello"`
Financial transactions	Transaction log ve reconciliation	`XADD transactions * amount "100.50"`
Game events	Player actions ve game state changes	`XADD game:events * action "level_up"`
Log aggregation	Centralized logging from multiple services	`XADD logs * service "api" level "error"`
CDC (Change Data Capture)	Database change tracking	`XADD changes * table "users" op "UPDATE"`

Stream'in en güçlü özelliklerinden biri, consumer groups mekanizmasıdır. Bu özellik sayesinde multiple consumers arasında load balancing, message acknowledgment ve redelivery işlemlerini otomatik olarak yönetebilirsiniz. Consumer group'lar, distributed processing, fault tolerance ve horizontal scaling için kritik öneme sahiptir. Bir consumer crash olsa bile, acknowledge edilmemiş message'lar otomatik olarak diğer consumer'lara redirect edilir.
Stream'in bir diğer devrimsel özelliği ise time-based range queries'dir. XRANGE ve XREVRANGE komutları ile belirli zaman aralıklarındaki message'ları millisecond precision ile çekebilirsiniz. Bu özellik, time-series analytics, historical data analysis ve event replay senaryoları için muazzam avantaj sağlar. Örneğin, "son 1 saatteki tüm error eventleri" veya "belirli user'ın bugünkü tüm aktiviteleri" gibi sorguları nanosecond seviyesinde hızla çözebilirsiniz.
🧩 1. Stream Veri Türü Temelleri
Redis'teki Stream, time-ordered, append-only log yapısıdır. Her entry (message) bir unique ID ve field-value çiftlerinden oluşur. Message ID'ler otomatik olarak timestamp-sequence formatında generate edilir ve chronological ordering garanti edilir. Stream'ler immutable'dır, yani eklenen message'lar sonradan değiştirilemez.
🔧 Temel Özellikler
Append-only: Sadece ekleme yapılabilir, modification mümkün değil
Unique IDs: Her message benzersiz timestamp-sequence ID alır
Time-ordered: Message'lar kronolojik sırada tutulur
Persistent: Message'lar disk'te kalıcı olarak saklanır
Field-value structure: Her message birden fazla field içerebilir

🔧 Temel Stream Komutları ve Açıklamaları

Komut	Açıklama	Örnek
`XADD`	Stream'e yeni message ekler	`XADD events * type "login"`
`XREAD`	Stream'den message'ları okur	`XREAD STREAMS events 0`
`XRANGE`	Belirli ID range'indeki message'ları getirir	`XRANGE events - +`
`XREVRANGE`	Reverse chronological order'da message'lar	`XREVRANGE events + - COUNT 10`
`XLEN`	Stream'deki message sayısını verir	`XLEN events`
`XGROUP`	Consumer group'ları yönetir	`XGROUP CREATE events group1 $`
`XREADGROUP`	Consumer group ile message okur	`XREADGROUP GROUP group1 c1`
`XACK`	Message'ı acknowledge eder	`XACK events group1 msg-id`
`XPENDING`	Acknowledge bekleyen message'ları listeler	`XPENDING events group1`
`XTRIM`	Stream'i belirli boyuta kırpar	`XTRIM events MAXLEN 1000`
`XINFO`	Stream hakkında bilgi verir	`XINFO STREAM events`
`XDEL`	Message'ları siler	`XDEL events message-id`

🧪 XADD Komutunu İnceleyelim

Stream'e yeni message eklemek için kullanılır.

text
127.0.0.1:6379> XADD events * event_type "user_login" user_id "123" timestamp "2024-01-20T10:30:00Z"  // Events stream'ine login eventi ekliyoruz.
"1642678200000-0"                                         // Otomatik generate edilen unique message ID.
127.0.0.1:6379> XADD events * event_type "page_view" user_id "456" page "/homepage"  // Page view eventi ekliyoruz.
"1642678201000-0"                                         // Yeni message ID (timestamp arttı).
127.0.0.1:6379> XADD events 1642678202000-0 event_type "purchase" user_id "123" amount "99.99"  // Manuel ID ile message ekliyoruz.
"1642678202000-0"                                         // Belirtilen ID kullanıldı.
127.0.0.1:6379> XADD metrics * cpu_usage "75.5" memory_usage "60.2" disk_usage "45.8"  // Metrics stream'ine sistem bilgileri.
"1642678203000-0"                                         // Metrics için yeni message ID.
127.0.0.1:6379> XADD chat:room1 * user "alice" message "Hello everyone!" timestamp "2024-01-20T10:35:00Z"  // Chat message'ı.
"1642678500000-0"                                         // Chat stream'inde message ID.

📌 Dikkat Edilmesi Gereken Noktalar
Otomatik ID generation * kullandığınızda Redis otomatik olarak current timestamp ile unique ID generate eder.

Manuel ID specification

text
127.0.0.1:6379>
127.0.0.1:6379> XADD events 1000000000000-0 type "test"   // Manuel olarak ID belirtiyoruz.
127.0.0.1:6379>

ID format: timestamp-sequence Format: millisecond_timestamp-sequence_number

Field-value çiftleri Her message birden fazla field-value çifti içerebilir:

text
127.0.0.1:6379>
127.0.0.1:6379> XADD orders * customer_id "C123" product "laptop" quantity "2" total "2999.98"
127.0.0.1:6379>

Stream yoksa otomatik oluşturulur İlk XADD komutu ile stream otomatik olarak oluşturulur.

🧪 XREAD Komutunu İnceleyelim

Stream'den message'ları okumak için kullanılır.

bash
127.0.0.1:6379> XREAD STREAMS events 0                   // events stream'ini baştan okuyoruz.
1) 1) "events"                                            // Stream adı.
   2) 1) 1) "1642678200000-0"                             // Message ID.
         2) 1) "event_type"                               // Field 1.
            2) "user_login"                               // Value 1.
            3) "user_id"                                  // Field 2.
            4) "123"                                      // Value 2.
            5) "timestamp"                                // Field 3.
            6) "2024-01-20T10:30:00Z"                     // Value 3.
      2) 1) "1642678201000-0"                             // İkinci message ID.
         2) 1) "event_type"
            2) "page_view"
            3) "user_id"
            4) "456"
            5) "page"
            6) "/homepage"
127.0.0.1:6379> XREAD COUNT 1 STREAMS events 0           // Sadece 1 message okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678200000-0"
         2) 1) "event_type"
            2) "user_login"
            3) "user_id"
            4) "123"
            5) "timestamp"
            6) "2024-01-20T10:30:00Z"
127.0.0.1:6379> XREAD BLOCK 5000 STREAMS events $        // 5 saniye boyunca yeni message'ları bekle.
(nil)                                                     // 5 saniyede yeni message gelmedi.
127.0.0.1:6379> XREAD STREAMS events metrics 0 0         // Birden fazla stream'i aynı anda okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678200000-0"
         2) 1) "event_type"
            2) "user_login"
            ... (events stream messages)
2) 1) "metrics"
   2) 1) 1) "1642678203000-0"
         2) 1) "cpu_usage"
            2) "75.5"
            ... (metrics stream messages)

📌 Dikkat Edilmesi Gereken Noktalar
Starting ID parametresi 0 = en baştan oku, $ = en son message'dan sonra oku (yeni message'ları bekle).

BLOCK seçeneği ile real-time reading

bash
127.0.0.1:6379>
127.0.0.1:6379> XREAD BLOCK 0 STREAMS events $           // Sonsuza kadar yeni message bekle.
127.0.0.1:6379>

COUNT ile limit setting

text
127.0.0.1:6379>
127.0.0.1:6379> XREAD COUNT 5 STREAMS events 0           // Maksimum 5 message oku.
127.0.0.1:6379>

Multiple stream reading Birden fazla stream'i aynı anda monitor edebilirsiniz.

🧪 XRANGE ve XREVRANGE Komutlarını İnceleyelim

Belirli ID aralığındaki message'ları okur.

text
127.0.0.1:6379> XRANGE events - +                        // Tüm message'ları kronolojik sırada getir.
1) 1) "1642678200000-0"
   2) 1) "event_type"
      2) "user_login"
      3) "user_id"
      4) "123"
      5) "timestamp"
      6) "2024-01-20T10:30:00Z"
2) 1) "1642678201000-0"
   2) 1) "event_type"
      2) "page_view"
      3) "user_id"
      4) "456"
      5) "page"
      6) "/homepage"
127.0.0.1:6379> XRANGE events 1642678200000 1642678201000  // Belirli timestamp aralığı.
1) 1) "1642678200000-0"
   2) 1) "event_type"
      2) "user_login"
      ... (message details)
2) 1) "1642678201000-0"
   2) 1) "event_type"
      2) "page_view"
      ... (message details)
127.0.0.1:6379> XRANGE events - + COUNT 2                // İlk 2 message'ı getir.
1) 1) "1642678200000-0"
   ... (first message)
2) 1) "1642678201000-0"
   ... (second message)
127.0.0.1:6379> XREVRANGE events + - COUNT 1             // En son message'ı getir (reverse order).
1) 1) "1642678202000-0"
   2) 1) "event_type"
      2) "purchase"
      3) "user_id"
      4) "123"
      5) "amount"
      6) "99.99"

📌 Dikkat Edilmesi Gereken Noktalar
Special boundaries - = stream'in başlangıcı, + = stream'in sonu.

Timestamp-based queries

text
127.0.0.1:6379>
127.0.0.1:6379> XRANGE events (1642678200000 1642678202000  // Exclusive range (1642678200000 hariç).
127.0.0.1:6379>

COUNT limitation Performance için COUNT kullanarak result set'i sınırlayın.
XREVRANGE reverse order En yeni message'lardan geriye doğru okuma için kullanın.

🧪 Consumer Groups - XGROUP ve XREADGROUP

Distributed message processing için consumer group'lar oluşturur.

bash
127.0.0.1:6379> XGROUP CREATE events processors $ MKSTREAM  // "processors" consumer group'u oluşturuyoruz.
OK                                                        // Group başarıyla oluşturuldu.
127.0.0.1:6379> XADD events * task_type "send_email" recipient "user@example.com" priority "high"  // Yeni task ekliyoruz.
"1642678300000-0"
127.0.0.1:6379> XREADGROUP GROUP processors worker1 COUNT 1 STREAMS events >  // worker1 olarak message okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678300000-0"
         2) 1) "task_type"
            2) "send_email"
            3) "recipient"
            4) "user@example.com"
            5) "priority"
            6) "high"
127.0.0.1:6379> XADD events * task_type "generate_report" user_id "789" format "pdf"  // Başka bir task.
"1642678301000-0"
127.0.0.1:6379> XREADGROUP GROUP processors worker2 COUNT 1 STREAMS events >  // worker2 olarak okuyoruz.
1) 1) "events"
   2) 1) 1) "1642678301000-0"
         2) 1) "task_type"
            2) "generate_report"
            3) "user_id"
            4) "789"
            5) "format"
            6) "pdf"
127.0.0.1:6379> XACK events processors 1642678300000-0   // worker1'in task'ını acknowledge ediyoruz.
(integer) 1                                               // 1 message acknowledge edildi.
127.0.0.1:6379> XPENDING events processors               // Pending (acknowledge edilmemiş) message'ları kontrol ediyoruz.
1) (integer) 1                                            // 1 pending message var.
2) "1642678301000-0"                                     // En eski pending message ID.
3) "1642678301000-0"                                     // En yeni pending message ID.
4) 1) 1) "worker2"                                        // Pending message'ı alan consumer.
      2) "1"                                             // Bu consumer'ın pending message sayısı.

📌 Dikkat Edilmesi Gereken Noktalar
Consumer group creation $ ile group'u stream'in en sonundan başlatıyoruz (sadece yeni message'ları okur).
Multiple workers Aynı group'taki farklı consumer'lar farklı message'ları alır (load balancing).
Message acknowledgment XACK ile message'ın işlendiğini belirtmek zorundasınız.
Pending messages Acknowledge edilmemiş message'lar XPENDING ile takip edilir.

Consumer failure handling

text
127.0.0.1:6379>
127.0.0.1:6379> XCLAIM events processors worker1 300000 1642678301000-0  // 5 dakika sonra message'ı başka worker'a transfer et.
127.0.0.1:6379>

🧪 XLEN ve XTRIM Komutlarını İnceleyelim

Stream bilgileri ve memory management için kullanılır.

text
127.0.0.1:6379> XLEN events                              // events stream'indeki message sayısı.
(integer) 4                                               // 4 message var.
127.0.0.1:6379> XADD events * test_type "load_test"      // Load test için çok sayıda message ekliyoruz.
127.0.0.1:6379> XADD events * test_type "load_test"
127.0.0.1:6379> XADD events * test_type "load_test"
127.0.0.1:6379> XLEN events                              // Güncel message sayısı.
(integer) 7
127.0.0.1:6379> XTRIM events MAXLEN 5                    // Stream'i 5 message ile sınırla.
(integer) 2                                               // 2 eski message silindi.
127.0.0.1:6379> XLEN events                              // Trimming sonrası message sayısı.
(integer) 5
127.0.0.1:6379> XTRIM events MAXLEN ~ 3                  // Approximately 3 message kalsın (~: yaklaşık).
(integer) 2                                               // 2 message daha silindi.
127.0.0.1:6379> XLEN events
(integer) 3

📌 Dikkat Edilmesi Gereken Noktalar
MAXLEN ile exact trimming Tam olarak belirtilen sayıda message kalır.
~ ile approximate trimming Performance için yaklaşık trimming yapar (daha hızlı).

Automatic trimming

text
127.0.0.1:6379>
127.0.0.1:6379> XADD events MAXLEN ~ 1000 * type "auto_trim"  // Add sırasında otomatik trim.
127.0.0.1:6379>

Memory optimization Büyük stream'ler için düzenli trimming yapın.
🔄 Stream ile Gerçek Dünya Senaryoları

📊 Senaryo 1: E-commerce Order Processing Pipeline

bash
127.0.0.1:6379> XGROUP CREATE orders order_processors $ MKSTREAM  // Order processing group'u oluşturuyoruz.
OK
127.0.0.1:6379> XADD orders * order_id "ORD001" customer_id "C123" total "299.99" status "pending"  // Yeni order.
"1642678400000-0"
127.0.0.1:6379> XADD orders * order_id "ORD002" customer_id "C456" total "149.50" status "pending"
"1642678401000-0"
127.0.0.1:6379> XREADGROUP GROUP order_processors payment_service COUNT 1 STREAMS orders >  // Payment service order alıyor.
1) 1) "orders"
   2) 1) 1) "1642678400000-0"
         2) 1) "order_id"
            2) "ORD001"
            3) "customer_id"
            4) "C123"
            5) "total"
            6) "299.99"
            7) "status"
            8) "pending"
# Payment service order'ı işledikten sonra:
127.0.0.1:6379> XACK orders order_processors 1642678400000-0  // Order acknowledge ediliyor.
(integer) 1
127.0.0.1:6379> XADD orders * order_id "ORD001" customer_id "C123" total "299.99" status "paid"  // Payment completed eventi.
"1642678402000-0"
127.0.0.1:6379> XREADGROUP GROUP order_processors inventory_service COUNT 1 STREAMS orders >  // Inventory service sıradaki order'ı alıyor.
1) 1) "orders"
   2) 1) 1) "1642678401000-0"
         2) 1) "order_id"
            2) "ORD002"
            ... (order details)
127.0.0.1:6379> XINFO GROUPS orders                      // Consumer group bilgilerini kontrol ediyoruz.
1) 1) "name"
   2) "order_processors"
   3) "consumers"
   4) (integer) 2                                         // 2 aktif consumer.
   5) "pending"
   6) (integer) 1                                         // 1 pending message.

🔥 Senaryo 2: Real-time IoT Sensor Data Collection

bash
127.0.0.1:6379> XGROUP CREATE sensor_data data_processors $ MKSTREAM  // Sensor data processing group.
OK
127.0.0.1:6379> XADD sensor_data * sensor_id "TEMP001" location "warehouse_a" temperature "23.5" humidity "65.2"  // Sıcaklık sensörü.
"1642678500000-0"
127.0.0.1:6379> XADD sensor_data * sensor_id "PRES001" location "tank_1" pressure "145.7" unit "psi"  // Basınç sensörü.
"1642678501000-0"
127.0.0.1:6379> XADD sensor_data * sensor_id "TEMP001" location "warehouse_a" temperature "24.1" humidity "66.8"  // Temp update.
"1642678502000-0"
# Real-time monitoring:
127.0.0.1:6379> XREAD BLOCK 1000 STREAMS sensor_data $   // 1 saniye boyunca yeni sensor data bekle.
# Analytics service specific sensor'ları okur:
127.0.0.1:6379> XREADGROUP GROUP data_processors analytics_worker COUNT 2 STREAMS sensor_data >
1) 1) "sensor_data"
   2) 1) 1) "1642678500000-0"
         2) 1) "sensor_id"
            2) "TEMP001"
            ... (sensor data)
      2) 1) "1642678501000-0"
         2) 1) "sensor_id"
            2) "PRES001"
            ... (pressure data)
# Time-range analytics:
127.0.0.1:6379> XRANGE sensor_data 1642678500000 1642678502000  // Son 2 saniyedeki tüm sensor readings.
1) 1) "1642678500000-0"
   ... (temp data 1)
2) 1) "1642678501000-0"
   ... (pressure data)
3) 1) "1642678502000-0"
   ... (temp data 2)
127.0.0.1:6379> XACK sensor_data data_processors 1642678500000-0 1642678501000-0  // Batch acknowledge.
(integer) 2

💬 Senaryo 3: Real-time Chat Application

markdown
127.0.0.1:6379> XADD chat:room1 * user_id "alice" username "Alice" message "Hello everyone!" timestamp "2024-01-20T11:00:00Z"  // Chat message.
"1642680000000-0"
127.0.0.1:6379> XADD chat:room1 * user_id "bob" username "Bob" message "Hi Alice!" timestamp "2024-01-20T11:00:15Z"
"1642680015000-0"
127.0.0.1:6379> XADD chat:room1 * user_id "charlie" username "Charlie" message "Good morning!" timestamp "2024-01-20T11:00:30Z"
"1642680030000-0"
# Real-time message listening:
127.0.0.1:6379> XREAD BLOCK 0 STREAMS chat:room1 1642680030000-0  // Son message'dan sonraki yeni message'ları bekle.
# Message history for new user:
127.0.0.1:6379> XREVRANGE chat:room1 + - COUNT 10        // Son 10 message'ı getir (history).
1) 1) "1642680030000-0"
   2) 1) "user_id"
      2) "charlie"
      3) "username"
      4) "Charlie"
      5) "message"
      6) "Good morning!"
      7) "timestamp"
      8) "2024-01-20T11:00:30Z"
2) 1) "1642680015000-0"
   ... (Bob's message)
3) 1) "1642680000000-0"
   ... (Alice's message)
# Message count:
127.0.0.1:6379> XLEN chat:room1                          // Toplam message sayısı.
(integer) 3
# Auto-cleanup (retain last 100 messages):
127.0.0.1:6379> XTRIM chat:room1 MAXLEN ~ 100            // Son 100 message'ı tut.
(integer) 0                                               // Silinecek message yok (henüz 3 tane var).

🏦 Senaryo 4: Financial Transaction Audit Trail

bash
127.0.0.1:6379> XGROUP CREATE transactions audit_processors $ MKSTREAM  // Audit processing group.
OK
127.0.0.1:6379> XADD transactions * transaction_id "TXN001" account_from "ACC123" account_to "ACC456" amount "1000.00" type "transfer"  // Transfer transaction.
"1642680100000-0"
127.0.0.1:6379> XADD transactions * transaction_id "TXN002" account "ACC789" amount "500.00" type "deposit" source "external"
"1642680101000-0"
127.0.0.1:6379> XADD transactions * transaction_id "TXN003" account "ACC123" amount "150.00" type "withdrawal" location "ATM_001"
"1642680102000-0"
# Fraud detection service:
127.0.0.1:6379> XREADGROUP GROUP audit_processors fraud_detector COUNT 5 STREAMS transactions >
1) 1) "transactions"
   2) 1) 1) "1642680100000-0"
         2) 1) "transaction_id"
            2) "TXN001"
            ... (transaction details)
      2) 1) "1642680101000-0"
         ... (deposit transaction)
      3) 1) "1642680102000-0"
         ... (withdrawal transaction)
# Account-specific transaction history:
127.0.0.1:6379> XRANGE transactions - +                  // Tüm transactions (audit için).
# Compliance reporting (daily transactions):
127.0.0.1:6379> XRANGE transactions 1642636800000 1642723199999  // Belirli gün aralığındaki transactions.
127.0.0.1:6379> XACK transactions audit_processors 1642680100000-0 1642680101000-0 1642680102000-0  // Audit complete.
(integer) 3
127.0.0.1:6379> XINFO STREAM transactions                // Transaction stream bilgileri.
1) "length"
2) (integer) 3
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1642680102000-0"
9) "groups"
10) (integer) 1                                          // 1 consumer group.

Redis Stream veri türü, event-driven architectures ve real-time data processing konusunda Redis'in en gelişmiş ve güçlü araçlarından biridir. Append-only log yapısı, consumer groups ve time-based queries ile modern application patterns'larını native olarak destekler. Bu bölümde XADD, XREAD, XRANGE, XGROUP, XREADGROUP gibi temel komutları ve gerçek dünya senaryolarında Stream kullanımını detaylı örneklerle öğrendik. Stream'in append-only structure, consumer groups, time-based queries ve message acknowledgment özellikleri özellikle:
- Event sourcing ve CQRS patterns
- Message queues ve async processing
- Real-time analytics ve monitoring
- IoT data collection ve processing
- Chat applications ve real-time communication
- Financial transaction logging
- Audit trails ve compliance gibi senaryolarda vazgeçilmez bir araçtır.
Stream veri türü, event-driven applications ihtiyaçlarınız için Redis'te en comprehensive çözümü sunar. Time-ordered persistence, distributed processing ve real-time capabilities üçlüsü, modern applications'ların complex event handling requirements'larını karşılamak için mükemmel bir kombinasyon oluşturur. Bu bölümde öğrendiğiniz komutlar ve senaryolar ile artık:
Event sourcing ve CQRS pattern implementasyonları geliştirebilir
Message queue sistemleri ve async processing pipelines tasarlayabilir
Real-time data streaming applications oluşturabilir
IoT data collection ve sensor monitoring sistemleri geliştirebilir
Audit trail ve transaction logging mekanizmaları implementasyonları yapabilirsiniz Stream'leri projelerinizde etkin kullanarak, kullanıcılarınıza event-driven, real-time ve highly-scalable deneyimler sunabilirsiniz. Sırada ne var? Bir sonraki yazımızda Redis'in Geo veri türünü derinlemesine inceleyeceğiz. Geo, geospatial indexing ve location-based queries için tasarlanmış specialized bir veri türüdür. GPS coordinates'leri nasıl verimli şekilde saklayabileceğimizi, proximity searches nasıl yapabileceğimizi, radius-based queries ile location services nasıl geliştirebileceğimizi ve geospatial applications'ın mobile era'daki kritik önemini öğreneceğiz.
Konuya başlamadan önce çok fazla özel isim kullandığımı fark ettim. Bu konuda elimden geldiğince sizlere tek tek bu kavramları öncelikle açıklamak istiyorum.
📝 Ek Bilgi: Önemli Kavramlar
📚 Temel Yapı Kavramları 📝 Append-only Log
Append-only = Sadece ekleme yapılabilir, silme/değiştirme olmaz
Log = Kronolojik sırayla tutulan kayıt defteri
Örnek: Banka hesap hareketleri gibi → Her işlem eklenir, değiştirilmez
Avantaj: Veri bütünlüğü ve audit trail garantisi
🔗 Chronological Ordering
Chronological = Zaman sırasına göre dizilmiş
Ordering = Sıralama sistemi
Redis'te her mesaj zaman damgası alır ve kronolojik sırada tutulur
Böylece "10:30'daki mesajlar" veya "bugünkü tüm olaylar" sorgulanabilir
🆔 Unique Message IDs
Her mesaj benzersiz ID alır: 1642678200000-0 formatında
İlk kısım: Timestamp (Unix millisecond)
İkinci kısım: Sequence number (aynı millisecond'deki sıra)
Hiçbir mesaj aynı ID'ye sahip olamaz
🏗️ Field-value Çiftleri
Her mesaj anahtar-değer çiftlerinden oluşur
Örnek: user_id: "123", action: "login", time: "10:30"
JSON'a benzer ama daha hızlı ve Redis'e optimize
🔒 Immutable
Immutable = Değiştirilemez, sabit
Mesaj eklendikten sonra içeriği değiştirilemez
Sadece XDEL ile silinebilir
Veri güvenliği ve tutarlılık için kritik
⏱️ Timestamp-sequence
Timestamp: Unix zaman damgası (millisecond)
Sequence: Aynı anda gelen mesajları ayırt etmek için
Format: <timestamp>-<sequence>
Otomatik generate edilir veya manuel verilebilir
🔄 İşlemci/Consumer Kavramları
👥 Consumer Groups
Consumer = Mesaj işleyen uygulama/servis
Group = Birlikte çalışan consumer'lar topluluğu
Her mesaj sadece grup içindeki 1 consumer tarafından işlenir
Load balancing ve parallel processing sağlar
⚖️ Load Balancing
Load = İş yükü/veri miktarı
Balancing = Dengeleme/dağıtma
Mesajlar consumer'lar arasında eşit dağıtılır
1000 mesaj + 5 consumer = Her biri ~200 mesaj alır
🔄 Redelivery
Redelivery = Yeniden teslim etme
Consumer crash olursa, işlenmemiş mesajlar başka consumer'a verilir
Veri kaybı olmaz, güvenilirlik artar
Otomatik failover mekanizması
🌐 Distributed Processing
Distributed = Dağıtık/çoklu makine
Processing = İşleme/hesaplama
Mesajlar farklı sunuculardaki consumer'lar tarafından işlenir
Horizontal scaling ve performance artışı
📈 Horizontal Scaling
Horizontal = Yatay (makine sayısı artırma)
Scaling = Ölçeklendirme/büyütme
Daha fazla consumer ekleyerek kapasiteyi artırma
Vertical scaling'in tersi (tek makinede RAM/CPU artırma)
🛡️ Fault Tolerance
Fault tolerance (hata toleransı), sistemin bir veya birden fazla bileşeni arızalandığında bile çalışmaya devam etme yeteneğidir
Redis Stream'lerde, bir consumer crash olsa bile diğer consumer'lar işi devralabilir
Örnek: E-ticaret siparişi işleyen 3 consumer var → 1'i çökerse → kalan 2'si işleme devam eder
Bu sayede sistem 7/24 kesintisiz çalışır ve veri kaybı olmaz
⏰ Zaman Tabanlı Kavramlar
🔍 Time-based Range Queries
Time-based = Zaman temelli
Range = Aralık/başlangıç-bitiş
Queries = Sorgular
Belirli zaman aralığındaki mesajları getirme
Örnek: "14:00-15:00 arası tüm error'lar"
📊 Time-series Analytics
Time-series = Zaman serisi verileri
Analytics = Analiz/çözümleme
Zamana göre değişen verilerin analizi
Örnek: CPU kullanımı, satış rakamları, sensör verileri
📚 Historical Data Analysis
Historical = Geçmiş/tarihsel
Data Analysis = Veri analizi
Geçmiş verileri inceleyip pattern bulma
Örnek: "Geçen ay en çok hangi saatlerde login olmuş?"
🔄 Event Replay
Event = Olay/durum değişikliği
Replay = Yeniden oynatma
Geçmiş olayları tekrar işleme/simülasyon
Debugging ve testing için kritik
✅ Mesaj Durumu Kavramları
✅ Message Acknowledgment
Message acknowledgment (mesaj onaylama), bir consumer'ın mesajı başarıyla işlediğini bildirmesidir
Consumer mesajı aldıktan sonra XACK komutu ile "işledim" der
Eğer consumer crash olur ve acknowledge etmezse, mesaj pending (beklemede) kalır
Pending mesajlar otomatik olarak diğer consumer'lara redelivery (yeniden teslim) edilir
⏳ Pending
Pending = Beklemede/askıda
Consumer'a verilmiş ama henüz acknowledge edilmemiş mesajlar
Consumer crash olursa bu mesajlar diğer consumer'lara verilir
XPENDING komutu ile görülebilir
💾 Kalıcılık Kavramları
💿 Persistent Storage
Persistent = Kalıcı/sürekli
Storage = Depolama
Mesajlar RAM'de değil disk'te saklanır
Server restart olsa bile mesajlar korunur
🗂️ Persistent Message Queues
Message Queues = Mesaj kuyrukları
Disk'te kalıcı olarak saklanan mesaj kuyrukları
Reliability ve durability garantisi
Elektrik kesilse bile mesajlar kaybolmaz
🏃‍♂️ Real-time Kavramları
⚡ Real-time Analytics
Real-time = Gerçek zamanlı/anlık
Analytics = Analiz/çözümleme
Veriler gelir gelmez anında işleme
Millisecond seviyesinde hızlı yanıt
📞 Blocking Operations
Blocking = Bekletme/engelleyici
Operations = İşlemler
Yeni mesaj gelene kadar bekleme yapma
XREAD BLOCK komutu ile gerçek zamanlı dinleme
🎯 Uygulama Alanları
📝 Event Sourcing
Event Sourcing = Olay kaynağı
Sistemdeki her değişikliği event olarak saklama
State yerine events tutma
Audit, replay ve debugging avantajları
📨 Message Queues
Message = Mesaj/veri paketi
Queues = Kuyruklar
Asynchronous (eş zamanlı olmayan) işlem kuyruğu
Producer-Consumer pattern
🏭 IoT Data Collection
IoT = Internet of Things (Nesnelerin İnterneti)
Data Collection = Veri toplama
Sensör verilerinin sürekli toplanması
Temperature, humidity, motion verileri
🧾 Audit Trails
Audit = Denetim/kontrol
Trails = İzler/kayıtlar
Sistem üzerindeki tüm işlemlerin kaydı
Compliance ve security için kritik
💰 Financial Transactions
Financial = Finansal/parasal
Transactions = İşlemler
Para transferi, ödeme kayıtları
ACID properties ve integrity garantisi
🔄 CDC (Change Data Capture)
Change Data Capture = Veri değişikliği yakalama
Database'deki değişiklikleri anlık yakalama
INSERT, UPDATE, DELETE operasyonları
Data synchronization için kullanılır

Redis Stream Veri Türü

📋 Redis Stream Veri Türü Özellikleri Tablosu

🎯 Redis Stream Kullanım Senaryoları

🧩 1. Stream Veri Türü Temelleri

🔧 Temel Özellikler

🔧 Temel Stream Komutları ve Açıklamaları

🧪 XADD Komutunu İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XREAD Komutunu İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XRANGE ve XREVRANGE Komutlarını İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 Consumer Groups - XGROUP ve XREADGROUP

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XLEN ve XTRIM Komutlarını İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🔄 Stream ile Gerçek Dünya Senaryoları

📊 Senaryo 1: E-commerce Order Processing Pipeline

🔥 Senaryo 2: Real-time IoT Sensor Data Collection

💬 Senaryo 3: Real-time Chat Application

🏦 Senaryo 4: Financial Transaction Audit Trail

📝 Ek Bilgi: Önemli Kavramlar

Not içeriği yükleniyor...

Redis Stream Veri Türü

📋 Redis Stream Veri Türü Özellikleri Tablosu

🎯 Redis Stream Kullanım Senaryoları

🧩 1. Stream Veri Türü Temelleri

🔧 Temel Özellikler

🔧 Temel Stream Komutları ve Açıklamaları

🧪 XADD Komutunu İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XREAD Komutunu İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XRANGE ve XREVRANGE Komutlarını İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🧪 Consumer Groups - XGROUP ve XREADGROUP

📌 Dikkat Edilmesi Gereken Noktalar

🧪 XLEN ve XTRIM Komutlarını İnceleyelim

📌 Dikkat Edilmesi Gereken Noktalar

🔄 Stream ile Gerçek Dünya Senaryoları

📊 Senaryo 1: E-commerce Order Processing Pipeline

🔥 Senaryo 2: Real-time IoT Sensor Data Collection

💬 Senaryo 3: Real-time Chat Application

🏦 Senaryo 4: Financial Transaction Audit Trail

📝 Ek Bilgi: Önemli Kavramlar