Message Queues — Kafka আর RabbitMQ-র দুই দর্শন

"""minilog.py — single-node partitioned log broker (~120 line)"""
import hashlib

def partition_for(key, count):
    h = int(hashlib.md5(str(key).encode()).hexdigest(), 16)
    return h % count

class MiniLogBroker:
    def __init__(self, partitions=3):
        self.logs = [[] for _ in range(partitions)]   # append-only!
        self.groups = {}                              # g -> {pid: offset}

    def produce(self, key, value):
        p = partition_for(key, len(self.logs))
        self.logs[p].append((key, value))
        return p, len(self.logs[p])                   # (partition, next-offset)

    def offsets(self, group):
        return self.groups.setdefault(
            group, {i: 0 for i in range(len(self.logs))})

    def poll(self, group, max_records=4):
        """bounded pull — মোট max_records, partition ঘুরে ঘুরে"""
        offs = dict(self.offsets(group))
        batch, progress = [], True
        while progress and len(batch) != max_records:
            progress = False
            for pid in range(len(self.logs)):
                if len(batch) == max_records:
                    break
                start = offs[pid]
                if start >= len(self.logs[pid]):
                    continue
                batch.append((pid, start, self.logs[pid][start]))
                offs[pid] = start + 1
                progress = True
        return batch

    def commit(self, group, positions):               # consumer-ই progress-
        self.offsets(group).update(positions)         # এর একমাত্র মালিক

    def lag(self, group):
        offs = self.groups.get(group, {})
        return {pid: len(log) - offs.get(pid, 0)
                for pid, log in enumerate(self.logs)}

    def total_lag(self, group):
        return sum(self.lag(group).values())

    def seek(self, group, pid, offset):               # replay-API!
        self.offsets(group)[pid] = max(0, offset)


# ── benchmark: 1 fast + 3 slow consumer-groups — lag dynamics ──
def benchmark():
    broker = MiniLogBroker(partitions=3)
    TICKS, PRODUCE_PER_TICK = 30, 8
    SPEEDS = {'fast': 24, 'slow-1': 4, 'slow-2': 2, 'v-slow': 1}

    print(f"{'tick':>4} {'produced':>8} " +
          " ".join(f"{g:>7}" for g in SPEEDS))
    produced = 0
    for t in range(1, TICKS + 1):
        for i in range(PRODUCE_PER_TICK):             # steady traffic
            broker.produce(f'user-{i % 5}', f'e{t}-{i}')
        produced += PRODUCE_PER_TICK
        row = ''
        for g, speed in SPEEDS.items():
            batch = broker.poll(g, max_records=speed)  # pull, own pace
            pos = dict(broker.offsets(g))
            for pid, off, rec in batch:
                pos[pid] = off + 1                     # process → commit
            broker.commit(g, pos)
            row += f"{broker.total_lag(g):>7} "
        if t == 1 or t % 10 == 0:
            print(f"{t:>4} {produced:>8} {row}")

    p0, _ = broker.produce('user-99', 'a')             # same-key ordering
    ok, prev = True, p0
    for i in range(5):
        p, nxt = broker.produce('user-99', f'step{i}')
        ok = ok and p == p0 and nxt > prev
        prev = nxt
    print(f"\nsame-key -> same-partition, in-order: {ok}")

    before = broker.total_lag('v-slow')
    broker.seek('v-slow', 0, 0)                        # replay demo
    print(f"seek(v-slow, p0->0): lag {before} -> "
          f"{broker.total_lag('v-slow')} (replay-দায় idempotency)")


benchmark()

নমুনা output:

tick  produced    fast  slow-1  slow-2  v-slow
   1        8       0       4       6       7 
  10       80       0      40      60      70 
  20      160       0      80     120     140 
  30      240       0     120     180     210 

same-key -> same-partition, in-order: True
seek(v-slow, p0->0): lag 216 -> 246 (replay-দায় idempotency)

তিনটা পাঠ:

• Fast lag শূন্য-ই; slow-group-গুলোর lag rate×time-এ linear — production alert-threshold-এর গণিত এখানেই।
• Same-key test প্রমাণ করে hash-placement-ই ordering-এর মূলধন।
• Seek-পরবর্তী lag-লাফ = replay-এর দাম — পুরনো record আবার process হবে, effect অবশ্যই idempotent (আগের লেসনের echo)।

নিজে বাড়ান:

1. Rebalance simulate করুন — v-slow-এর এক partition অন্য group-member-কে দিন (assignment-map যোগ করে)
2. Compacted-log mode — একই key-র শেষ record ছাড়া বাকি purge; lag-ganit তাতে কেমন বদলায় দেখুন
3. Dead-letter path — malformed record আলাদা partition-এ quarantine