Dalam aplikasi web dan seluler modern, API menjadi tulang punggung komunikasi antara berbagai komponen, layanan, dan pengguna. Namun, seiring dengan meningkatnya penggunaan API, terdapat risiko kelebihan beban sistem, yang menyebabkan penurunan kinerja atau bahkan penghentian layanan. Salah satu cara paling efektif untuk menghindari masalah tersebut adalah dengan membatasi throughput API.

Pembatasan tarif mengacu pada praktik membatasi jumlah permintaan yang dapat dibuat oleh pengguna atau sistem ke API dalam jangka waktu tertentu, yang diukur dalam permintaan per detik atau per menit. Hal ini memastikan bahwa tidak ada pengguna atau klien yang membebani API, memungkinkan penggunaan wajar dan melindungi backend agar tidak dibanjiri lalu lintas yang berlebihan.

Dalam artikel ini, kita akan mempelajari berbagai strategi pembatasan tarif yang tersedia, kasus penggunaannya, dan praktik terbaik dalam menerapkannya guna melindungi API dari kelebihan beban.

Mengapa pembatasan laju API itu penting?

Pembatasan tingkat API sangat penting untuk:

1. Mencegah banjir berbahaya dan serangan penolakan layanan (DOS).
2. Pertahankan kinerja dan keandalan API.
3. Menjamin penggunaan wajar antar pengguna.
4. Hindari biaya tinggi yang terkait dengan penggunaan layanan cloud secara berlebihan.

Strategi Umum Pembatasan Tingkat API

Ada beberapa strategi pembatasan kecepatan yang dapat diterapkan di gateway API, penyeimbang beban, dll.

1. Memperbaiki batasan aliran jendela

Strategi ini melibatkan penetapan batas tetap pada jumlah permintaan yang diizinkan dalam jangka waktu tertentu, misalnya 100 permintaan per menit. Penghitung disetel ulang di akhir jendela. Kelemahan utama adalah kemungkinan terjadinya masalah “gemuruh kawanan”. Jika beberapa pengguna mencapai batasnya tepat sebelum jendela direset, sistem dapat mengalami lonjakan lalu lintas, yang berpotensi menyebabkan kelebihan beban.

import time
class FixedWindowRateLimiter:
    def __init__(self, limit, window_size):
        self.limit = limit
        self.window_size = window_size
        self.requests = []
    def is_allowed(self):
        current_time = time.time()
        self.requests = [req for req in self.requests if req > current_time - self.window_size]
        # Check if the number of requests in the current window exceeds the limit
        if len(self.requests) < self.limit:
            self.requests.append(current_time)
            return True
        else:
            return False
# Example usage
limiter = FixedWindowRateLimiter(limit=5, window_size=60)  # 5 requests per minute
for _ in range(7):
    if limiter.is_allowed():
        print("Request allowed")
    else:
        print("Rate limit exceeded")
    time.sleep(10)  # Sleep for 10 seconds between requests

2. Pembatasan aliran dengan jendela geser

Strategi ini mencoba memecahkan masalah “kawanan guntur” dengan memindahkan jendela secara dinamis berdasarkan stempel waktu permintaan.

Dalam pendekatan ini, jendela bergerak terus-menerus dan permintaan dihitung berdasarkan periode terkini, sehingga memungkinkan distribusi lalu lintas lebih lancar dan kecil kemungkinannya menyebabkan lonjakan tiba-tiba. Seorang pengguna diperbolehkan membuat 100 permintaan dalam periode 60 detik. Jika mereka membuat permintaan 30 detik yang lalu, mereka hanya dapat membuat 99 permintaan lagi dalam 30 detik berikutnya. Strategi ini sedikit lebih rumit untuk diterapkan dan dikelola dibandingkan dengan strategi jendela tetap.

import time
from collections import deque
class SlidingWindowRateLimiter:
    def __init__(self, limit, window_size):
        self.limit = limit
        self.window_size = window_size
        self.requests = deque()
    def is_allowed(self):
        current_time = time.time()
        while self.requests and self.requests[0] < current_time - self.window_size:
            self.requests.popleft()
        if len(self.requests) < self.limit:
            self.requests.append(current_time)
            return True
        else:
            return False
# Example usage
limiter = SlidingWindowRateLimiter(limit=5, window_size=60)  
for _ in range(7):
    if limiter.is_allowed():
        print("Request allowed")
    else:
        print("Rate limit exceeded")
    time.sleep(10)  # Sleep for 10 seconds between requests

3. Batasan tarif keranjang token

Token bucket adalah salah satu algoritma yang paling banyak digunakan. Dalam pendekatan ini, token dihasilkan pada tingkat bunga tetap dan disimpan dalam keranjang. Setiap permintaan menghapus token dari keranjang. Jika keranjang kosong, permintaan ditolak hingga token baru dibuat.

Algoritme ini memerlukan pelacakan token dan status bucket yang cermat dan dapat menimbulkan beberapa kompleksitas dalam implementasinya. Ini lebih fleksibel daripada jendela tetap atau geser dan memungkinkan lonjakan permintaan sambil menerapkan throughput maksimum dari waktu ke waktu.

import time
class TokenBucketRateLimiter:
    def __init__(self, rate, capacity):
        self.rate = rate  
        self.capacity = capacity  
        self.tokens = capacity
        self.last_checked = time.time()
    def is_allowed(self):
        current_time = time.time()
        elapsed = current_time - self.last_checked
        self.tokens += elapsed * self.rate
        if self.tokens > self.capacity:
            self.tokens = self.capacity
        self.last_checked = current_time
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        else:
            return False
# Example usage
limiter = TokenBucketRateLimiter(rate=1, capacity=5) 
for _ in range(7):
    if limiter.is_allowed():
        print("Request allowed")
    else:
        print("Rate limit exceeded")
    time.sleep(1)  # Sleep for 1 second between requests

4. Membatasi laju kebocoran bucket

Mirip dengan algoritma token bucket, model leaky bucket menerapkan throughput maksimum dengan mengontrol aliran permintaan melalui sistem.

Dalam model ini, permintaan ditambahkan ke “keranjang” pada tingkat yang bervariasi, namun ember bocor pada tingkat yang tetap. Jika ember meluap, permintaan lainnya ditolak. Strategi ini membantu memperlancar lalu lintas yang padat sekaligus memastikan bahwa permintaan diproses pada tingkat yang konsisten. Mirip dengan token bucket, penerapannya bisa jadi rumit, terutama untuk sistem dengan variabilitas tinggi dalam lalu lintas kueri.

import time
class LeakyBucketRateLimiter:
    def __init__(self, rate, capacity):
        self.rate = rate  
        self.capacity = capacity 
        self.water_level = 0  
        self.last_checked = time.time()
    def is_allowed(self):
        current_time = time.time()
        elapsed = current_time - self.last_checked
        self.water_level -= elapsed * self.rate
        if self.water_level < 0:
            self.water_level = 0
        self.last_checked = current_time
        if self.water_level < self.capacity:
            self.water_level += 1
            return True
        else:
            return False
# Example usage
limiter = LeakyBucketRateLimiter(rate=1, capacity=5) 
for _ in range(7):
    if limiter.is_allowed():
        print("Request allowed")
    else:
        print("Rate limit exceeded")
    time.sleep(1)  # Sleep for 1 second between requests

5. Pembatasan tarif berbasis IP

Dalam kebijakan ini, batasan tarif diterapkan berdasarkan alamat IP pengguna. Hal ini memastikan bahwa permintaan dari satu alamat IP dibatasi pada ambang batas tertentu. Pendekatan ini dapat dielakkan oleh pengguna yang menggunakan VPN atau proxy. Selain itu, hal ini dapat berdampak tidak adil terhadap pengguna yang berbagi alamat IP.

import time
class IpRateLimiter:
    def __init__(self, limit, window_size):
        self.limit = limit
        self.window_size = window_size
        self.ip_requests = {}
    def is_allowed(self, ip):
        current_time = time.time()
        if ip not in self.ip_requests:
            self.ip_requests[ip] = []
        self.ip_requests[ip] = [req for req in self.ip_requests[ip] if req > current_time - self.window_size]
        if len(self.ip_requests[ip]) < self.limit:
            self.ip_requests[ip].append(current_time)
            return True
        else:
            return False
# Example usage
limiter = IpRateLimiter(limit=5, window_size=60)  
for ip in ['192.168.1.1', '192.168.1.2']:
    for _ in range(7):
        if limiter.is_allowed(ip):
            print(f"Request from {ip} allowed")
        else:
            print(f"Rate limit exceeded for {ip}")
        time.sleep(10)  # Sleep for 10 seconds between requests

6. Pembatasan Tarif Berbasis Pengguna

Ini adalah kebijakan pembatasan tarif yang lebih personal, dimana batasan tersebut diterapkan pada masing-masing pengguna atau akun yang diautentikasi, bukan pada alamat IP mereka. Untuk pengguna yang diautentikasi, pembatasan tarif dapat dilakukan berdasarkan akun mereka (misalnya, melalui kunci API atau token OAuth).

import time
class UserRateLimiter:
    def __init__(self, limit, window_size):
        self.limit = limit
        self.window_size = window_size
        self.user_requests = {}
    def is_allowed(self, user_id):
        current_time = time.time()
        if user_id not in self.user_requests:
            self.user_requests[user_id] = []
        self.user_requests[user_id] = [req for req in self.user_requests[user_id] if req > current_time - self.window_size]
        if len(self.user_requests[user_id]) < self.limit:
            self.user_requests[user_id].append(current_time)
            return True
        else:
            return False
# Example usage
limiter = UserRateLimiter(limit=5, window_size=60) 
for user_id in ['user1', 'user2']:
    for _ in range(7):
        if limiter.is_allowed(user_id):
            print(f"Request from {user_id} allowed")
        else:
            print(f"Rate limit exceeded for {user_id}")
        time.sleep(10)  # Sleep for 10 seconds between requests

Praktik terbaik untuk menerapkan pembatasan tarif

• Gunakan respons kesalahan yang jelas, biasanya “429 permintaan terlalu banyak”.
• Batas tarif berdasarkan konteks dan faktor seperti peran pengguna, titik akhir API, atau tingkat langganan.
• Batasan terperinci pada tingkat yang berbeda (misalnya global, per pengguna, per IP) berdasarkan kebutuhan API.
• Catat dan pantau pembatasan tarif untuk mengidentifikasi pola potensi penyalahgunaan atau penyalahgunaan.
• Gunakan Redis atau solusi caching serupa untuk sistem yang sangat terdistribusi.
• Gunakan interval eksponensial untuk mencoba lagi dengan interval penundaan yang bertambah.

Kesimpulan

Pembatasan tingkat API adalah aspek penting dari manajemen API yang memastikan kinerja, keandalan, dan keamanan. Dengan memilih strategi yang tepat berdasarkan kebutuhan sistem dan memantau pola penggunaan secara efektif, kesehatan dan kinerja API dapat dipertahankan, bahkan selama lalu lintas padat. Pembatasan suku bunga bukan sekedar tindakan defensif; Ini adalah bagian integral dalam menciptakan layanan web yang skalabel dan tangguh.