Redis Streams হল Redis-এর একটি শক্তিশালী ডেটা স্ট্রাকচার যা message queueing, event streaming, এবং data processing এর জন্য ব্যবহৃত হয়। Redis Streams-এর মাধ্যমে আপনি একটি stream তৈরি করতে পারেন, যেখানে বিভিন্ন producers (পাবলিশার্স) ডেটা পাবলিশ করে এবং consumers (সাবস্ক্রাইবাররা) সেই ডেটা প্রসেস করে। Redis Streams খুব কার্যকরী হতে পারে যখন আপনি রিয়েল-টাইম ডেটা প্রসেসিং, অ্যাসিঙ্ক্রোনাস মেসেজিং বা লগিং সিস্টেম তৈরি করতে চান।
Redis Streams-এর ধারণা
Redis Streams একটি log-based data structure যা একটি ordered sequence of messages বা events ধারণ করে। প্রতিটি message বা event একটি unique ID দিয়ে চিহ্নিত হয়, যা time-based হয় এবং একটি message payload ধারণ করে। Redis Streams-এর মাধ্যমে আপনি একটি Producer-Consumer মডেল তৈরি করতে পারেন যেখানে একাধিক producer এবং consumer অ্যাপ্লিকেশন একযোগে ডেটা প্রক্রিয়া করে।
Redis Streams এর মূল কমান্ডগুলো হলো:
- XADD - নতুন একটি entry স্ট্রিমে যোগ করা।
- XREAD - স্ট্রিম থেকে ডেটা পড়া।
- XGROUP - Consumer group তৈরি করা।
- XACK - Consumer group থেকে ডেটা প্রক্রিয়াকরণের পরে ACK (Acknowledgement) করা।
- XDEL - স্ট্রিম থেকে ডেটা মুছে ফেলা।
Redis Streams-এর মাধ্যমে Data Processing Example
ধরা যাক, আপনি একটি Producer-Consumer মডেল তৈরি করতে চান যেখানে producer ডেটা পাবলিশ করবে এবং consumer সেই ডেটা প্রসেস করবে। নিচে Redis Streams ব্যবহার করে data processing এর একটি সাধারণ উদাহরণ দেয়া হল।
1. Producer (XADD command)
Producer সাধারণত ডেটা Streams-এ পাঠায়। এটি XADD কমান্ড ব্যবহার করে স্ট্রিমে নতুন entry যোগ করে।
Producer কোড (Python Example)
import redis
# Redis সার্ভারে কানেক্ট করা
r = redis.Redis(host='localhost', port=6379, db=0)
# নতুন ডেটা Streams-এ যোগ করা
stream_name = 'mystream'
message_id = r.xadd(stream_name, {'event': 'user_created', 'user_id': '1234', 'name': 'John Doe'})
print(f"Message added with ID: {message_id}")
- XADD কমান্ডে,
stream_nameএর মধ্যে নতুন event পাঠানো হয়, যা একটি key-value পেয়ার হিসেবে থাকে। - এখানে
event,user_idএবংnameপ্যারামিটার পাঠানো হচ্ছে।
2. Consumer (XREAD command)
Consumer XREAD কমান্ড ব্যবহার করে স্ট্রিম থেকে ডেটা পড়ে এবং প্রক্রিয়াকরণ করে। এটি XREAD এর মাধ্যমে নতুন ডেটা পাওয়া গেলে তা প্রসেস করে।
Basic Consumer Example (Python)
import redis
# Redis সার্ভারে কানেক্ট করা
r = redis.Redis(host='localhost', port=6379, db=0)
# স্ট্রিম থেকে ডেটা পড়া
stream_name = 'mystream'
while True:
# নতুন ডেটা আসলে XREAD কমান্ড ব্যবহার করে পড়া
messages = r.xread({stream_name: '0'}, count=1, block=0)
for message in messages:
message_id, message_data = message
print(f"New message from {stream_name}: {message_data}")
# এখানে ডেটা প্রসেস করতে পারেন
user_id = message_data.get(b'user_id').decode('utf-8')
name = message_data.get(b'name').decode('utf-8')
print(f"Processing user: {user_id}, Name: {name}")
# Acknowledge message after processing
r.xack(stream_name, 'mygroup', message_id)
XREADকমান্ডের মাধ্যমে consumer স্ট্রিম থেকে ডেটা পড়বে।block=0ব্যবহার করে লম্বা সময় ধরে ডেটা রিড করা যাবে।xackকমান্ডের মাধ্যমে, ডেটা প্রসেস করার পরে consumer তা ack করে দেয়, যাতে consumer group পরবর্তী বার্তা গ্রহণ করতে পারে।
3. Consumer Group (XGROUP command)
Redis Streams-এ Consumer Groups ব্যবহৃত হয়, যেখানে একাধিক consumer একসাথে স্ট্রিমের ডেটা প্রক্রিয়া করতে পারে। এটি workload distribution এবং fault tolerance নিশ্চিত করে।
Consumer Group তৈরি করা:
r.xgroup_create('mystream', 'mygroup', id='0', make_stream=True)
- এখানে
mystreamস্ট্রিমের জন্যmygroupনামে একটি consumer group তৈরি করা হয়েছে। id='0'দ্বারা বলা হচ্ছে যে গ্রুপটি সমস্ত স্ট্রিম ডেটা শুরু থেকে গ্রহণ করবে।make_stream=Trueমানে, যদি স্ট্রিমটি আগে থেকে না থাকে, তবে এটি তৈরি করা হবে।
Consumer Group থেকে ডেটা পড়া:
while True:
# গ্রুপের মাধ্যমে ডেটা রিড করা
messages = r.xreadgroup('mygroup', 'consumer1', {stream_name: '>'}, count=1, block=0)
for message in messages:
message_id, message_data = message
print(f"New message: {message_data}")
# এখানে ডেটা প্রসেস করতে পারেন
user_id = message_data.get(b'user_id').decode('utf-8')
name = message_data.get(b'name').decode('utf-8')
print(f"Processing user: {user_id}, Name: {name}")
# Acknowledge message after processing
r.xack(stream_name, 'mygroup', message_id)
- XREADGROUP কমান্ডের মাধ্যমে consumer group ব্যবহার করে ডেটা পড়া হয়।
'>' এর মাধ্যমে এটি নতুন ডেটা গ্রহণ করে, এবং প্রতিটি message প্রক্রিয়া করার পরেXACKকমান্ড ব্যবহার করে তা ack করা হয়।
4. Handling Stream Messages Efficiently
Stream Trim (Xtrim):
স্ট্রিমের আকার বড় হতে পারে, তাই আপনি পুরনো ডেটা মুছে ফেলতে XTRIM ব্যবহার করতে পারেন।
r.xtrim('mystream', minid='1000', approximate=False)
এটি mystream স্ট্রিম থেকে 1000 এর চেয়ে পুরনো ডেটা মুছে ফেলবে।
5. Stream Processing with Batching
স্ট্রিমের ডেটা ব্যাচ আকারে প্রক্রিয়া করতে পারেন। এতে একাধিক ডেটা একত্রে প্রসেস করা হয় এবং তা performance ভালো হতে পারে।
messages = r.xread({stream_name: '0'}, count=10, block=0)
for message in messages:
# এখানে ব্যাচ প্রসেসিং করা হবে
print(message)
Conclusion
Redis Streams একটি শক্তিশালী ডেটা স্ট্রাকচার যা মেসেজ কিউইং এবং রিয়েল-টাইম ডেটা প্রসেসিংয়ে ব্যবহৃত হয়। আপনি Producer-Consumer মডেল ব্যবহার করে স্ট্রিমের মাধ্যমে ডেটা পাঠাতে এবং গ্রহণ করতে পারেন। Consumer Groups ব্যবহার করলে একাধিক consumer একই স্ট্রিম থেকে ডেটা প্রসেস করতে পারে, যা কার্যকারিতা বৃদ্ধি এবং ফেইলওভার সুবিধা দেয়। Redis Streams পারফরম্যান্স অপটিমাইজেশন এবং ডেটা ম্যানিপুলেশন নিশ্চিত করার জন্য বিভিন্ন কৌশল ব্যবহার করা যায়, যেমন batch processing, stream trimming, এবং XACK.
Read more
