خط أنابيب Redis

العميل                   خادم Redis
  |                         |
  |----- SET key1 -------->|
  |<----- OK --------------|
  |                         |
  |----- SET key2 -------->|
  |<----- OK --------------|
  |                         |
  |----- SET key3 -------->|
  |<----- OK --------------|

العميل                   خادم Redis
  |                         |
  |----- SET key1 -------->|
  |----- SET key2 -------->|
  |----- SET key3 -------->|
  |                         |
  |<----- OK --------------|
  |<----- OK --------------|
  |<----- OK --------------|

النهج التقليدي:
- 100 أمر = 100 جولة = 100ms زمن انتقال شبكة
- بالإضافة إلى وقت معالجة Redis (افترض 0.1ms/أمر) = 10ms
- الوقت الإجمالي = 100ms + 10ms = 110ms

خط الأنابيب:
- 100 أمر = 1 جولة = 1ms زمن انتقال شبكة
- بالإضافة إلى وقت معالجة Redis = 10ms
- الوقت الإجمالي = 1ms + 10ms = 11ms

# استخدام وضع خط أنابيب redis-cli
echo -e "SET key1 value1\nSET key2 value2\nSET key3 value3" | redis-cli

# أو القراءة من ملف
cat commands.txt | redis-cli

import redis

r = redis.Redis(host='localhost', port=6379)

# إنشاء خط أنابيب
pipe = r.pipeline()

# إضافة أوامر إلى خط الأنابيب
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')

# تنفيذ خط الأنابيب
results = pipe.execute()
print(results)  # [True, True, True]

Jedis jedis = new Jedis("localhost", 6379);

Pipeline pipeline = jedis.pipelined();

pipeline.set("key1", "value1");
pipeline.set("key2", "value2");
pipeline.set("key3", "value3");

List<Object> results = pipeline.syncAndReturnAll();

const Redis = require('ioredis');
const redis = new Redis();

const pipeline = redis.pipeline();

pipeline.set('key1', 'value1');
pipeline.set('key2', 'value2');
pipeline.set('key3', 'value3');

const results = await pipeline.exec();

import redis

r = redis.Redis()

# إنشاء خط أنابيب مع تمكين المعاملة
pipe = r.pipeline(transaction=True)

pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.incr('counter')

# تنفيذ المعاملة
results = pipe.execute()

import redis

r = redis.Redis()
pipe = r.pipeline()

# تعيين 1000 مفتاح دفعة واحدة
for i in range(1000):
    pipe.set(f'key:{i}', f'value:{i}')

results = pipe.execute()
print(f'Set {len(results)} keys')

import redis

r = redis.Redis()
pipe = r.pipeline()

# الحصول على 1000 مفتاح دفعة واحدة
keys = [f'key:{i}' for i in range(1000)]
for key in keys:
    pipe.get(key)

values = pipe.execute()
print(f'Got {len(values)} values')

import redis

r = redis.Redis()
pipe = r.pipeline()

# حذف دفعات
keys = ['key1', 'key2', 'key3', 'key4', 'key5']
for key in keys:
    pipe.delete(key)

results = pipe.execute()
print(f'Deleted {sum(results)} keys')

import redis
import json

r = redis.Redis()
pipe = r.pipeline()

# استيراد بيانات من ملف JSON
with open('data.json', 'r') as f:
    data = json.load(f)

    for item in data:
        key = f"user:{item['id']}"
        value = json.dumps(item)
        pipe.set(key, value)

pipe.execute()
print('Import complete')

import redis

r = redis.Redis()
pipe = r.pipeline()

# تحديث عدادات دفعات
counters = {
    'article:1:views': 10,
    'article:2:views': 20,
    'article:3:views': 30,
}

for key, increment in counters.items():
    pipe.incrby(key, increment)

pipe.execute()
print('Counters updated')

import redis
import time

r = redis.Redis()

# بيانات الاختبار
n = 10000

# الطريقة 1: النهج العادي
start = time.time()
for i in range(n):
    r.set(f'key:{i}', f'value:{i}')
end = time.time()
print(f'Regular: {end - start:.2f} seconds')

# الطريقة 2: خط الأنابيب
start = time.time()
pipe = r.pipeline()
for i in range(n):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()
end = time.time()
print(f'Pipeline: {end - start:.2f} seconds')

Regular: 5.23 seconds
Pipeline: 0.51 seconds

تحسين الأداء: حوالي 10 أضعاف

# ❌ إرسال عدد كبير جدًا من الأوامر مرة واحدة
pipe = r.pipeline()
for i in range(100000):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()  # قد يستخدم الكثير من الذاكرة

# ✅ التنفيذ على دفعات
batch_size = 1000
for batch in range(0, 100000, batch_size):
    pipe = r.pipeline()
    for i in range(batch, batch + batch_size):
        pipe.set(f'key:{i}', f'value:{i}')
    pipe.execute()

import redis

r = redis.Redis()
pipe = r.pipeline()

pipe.set('key1', 'value1')
pipe.incr('key1')  # سيفشل (key1 ليس رقمًا)
pipe.set('key2', 'value2')

try:
    results = pipe.execute()
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f'Command {i} failed: {result}')
except Exception as e:
    print(f'Pipeline execution failed: {e}')

import redis

r = redis.Redis()

# استخدم المعاملة عند الحاجة للذرية
pipe = r.pipeline(transaction=True)

pipe.set('account:a', '100')
pipe.set('account:b', '50')

try:
    results = pipe.execute()
    print('Transaction executed successfully')
except Exception as e:
    print(f'Transaction failed: {e}')

import redis
import time

r = redis.Redis()

start = time.time()
pipe = r.pipeline()

for i in range(10000):
    pipe.set(f'key:{i}', f'value:{i}')

results = pipe.execute()
end = time.time()

print(f'Execution time: {end - start:.2f} seconds')
print(f'QPS: {10000 / (end - start):.0f}')

# خط الأنابيب يخزّن جميع الأوامر والنتائج مؤقتًا
# دفعة كبيرة جدًا تستخدم الكثير من الذاكرة

# الحل: تنفيذ على دفعات

# الأوامر في خط الأنابيب يمكن مقاطعتها من قبل عملاء آخرين
# استخدم المعاملات عندما تكون الذرية مطلوبة

# لا يمكن استخدام نتيجة أمر سابق في خط الأنابيب

# ❌ مثال غير صحيح
pipe = r.pipeline()
pipe.incr('counter')
# لا يمكن الحصول على نتيجة incr للأوامر اللاحقة
pipe.set('result', ???)  # لا يمكن استخدام نتيجة incr
pipe.execute()

# ✅ الحل: استخدم برامج Lua النصية