ساخت میکروسرویس با پایتون: استفاده از Django، RabbitMQ و Pika

در این مقاله ما یک مثال ساده Microservice با Python Django و Python Flask ایجاد خواهیم کرد. ما یاد خواهیم گرفت که چگونه با RabbitMQ یک معماری رویداد محور ایجاد کنیم. برنامه ها از Python Pika برای ارسال رویدادها به سایر برنامه ها استفاده می کنند و به نظر می رسد همه چیز به عنوان یک برنامه Monolithic کار می کند اما در واقع یک برنامه Microservices است.

میکروسرویس چیست؟

Microservice یک سبک معماری است که یک سیستم نرم افزاری پیچیده را به عنوان مجموعه ای از سرویس های متصل به هم آزاد می سازد که با هر یک از آنها با استفاده از یک استاندارد از پیش تعریف شده ارتباط برقرار می کند. حرکت از معماری یکپارچه به میکروسرویس‌ها، سیستم‌های نرم‌افزاری را مقیاس‌پذیر، قابل استقرار، قابل استفاده مجدد و انعطاف‌پذیر در برابر شکست می‌کند. هر میکروسرویس مستقل از یکدیگر اجرا می شود و داده های خود را حفظ می کند.

همانطور که در نمودار بالا نشان داده شده است، میکروسرویس‌ها با یکدیگر از طریق یک کانال ارتباطی یا پیام‌رسان مشابه به عنوان RabbitMQ یا Apache Kafka ارتباط برقرار می‌کنند.

رابیت‌ام‌کیو یک دستگاه میانجی یا سیستم صف پیام است. صف‌ها برای انتقال پیام‌ها بین برنامه‌ها، که به عنوان میکروسرویس‌ها هم شناخته می‌شوند، استفاده می‌شوند.

چرا از میکروسرویس‌ها استفاده می‌شود؟

یک دستگاه میانجی پیام به برنامه‌ها اجازه می‌دهد که بتوانند پیام‌ها را بفرستند و دریافت کنند بدون اینکه نیاز به آگاهی از وجود یا مکان یکدیگر داشته باشند. این امر به بهبود قابلیت مقیاس‌پذیری، پایداری و عملکرد کمک می‌کند. همچنین، دستگاه‌های میانجی پیام برای کاهش بار و زمان‌های تحویل سرورهای برنامه با اختصاص وظایف به برنامه‌های دیگر استفاده می‌شوند.

اصطلاحات مرتبط با این معماری

تولیدکننده (Producer): برنامه‌ای که پیام‌ها را ارسال می‌کند.

مصرف‌کننده (Consumer): برنامه‌ای که پیام‌ها را دریافت می‌کند.

صف (Queue): حافظه موقت برای ذخیره پیام‌ها.

تبادل (Exchange): به جای ارسال پیام مستقیماً به صف، تولیدکننده آن را به یک تبادل ارسال می‌کند. تبادل منطق مسیریابی را فراهم می‌کند تا تصمیم بگیرد پیام به کدام یک یا چند صف ارسال شود.

پیوند (Binding): ارتباط بین یک صف و یک تبادل.

کلید مسیریابی (Routing key): یک کلید استفاده شده توسط تبادل برای تصمیم گیری در مورد مسیریابی.

پروتکل AMQP (پروتکل پیام‌رسانی پیشرفته): پروتکل استفاده شده توسط RabbitMQ برای ارسال و دریافت پیام‌ها.

تبادل مستقیم (مسیریابی): در این حالت، یک پیام به صف‌هایی می‌رود که کلید پیوند آن‌ها با کلید مسیریابی پیام یکسان است. اهمیت دارد که توجه شود که یک صف ممکن است به چندین کلید پیوند متصل شود.

تبادل موضوعی: این مانند تبادل مستقیم است، با این تفاوت که به جای استفاده از یک کلید مسیریابی ثابت، از نمادهای ویژگی برای تعریف الگوهای مسیریابی استفاده می‌کند.

بیایید شروع کنیم

یک سیستم می‌خواهد ایمیل خوش‌آمدگویی به کاربرانی که در فرآیند ثبت‌نام شده‌اند ارسال کند و همچنین جزئیات ثبت‌نام را در سیستم لاگر ثبت کند. در معماری یکپارچه، ثبت‌نام کاربر، ارسال ایمیل و ثبت لاگ‌ها توسط یک عامل انجام می‌شود. این باید به یک سیستم با کارکرد‌های مستقل تبدیل شود که مسئولیت‌های ثبت‌نام کاربر، ارسال ایمیل و ثبت لاگ توسط مؤلفه‌های مختلف انجام شود. به عبارت دیگر، می‌خواهیم که برنامه‌ی خود را به 3 میکروسرویس تقسیم کنیم به شرح زیر:

1.هر زمان که کاربر جدیدی ایجاد می‌شود، سرویس کاربر(UserService) یک پیام با کلید مسیریابی “user.created.key” و داده‌های JSON حاوی اطلاعات کاربر به صورت {“id”: 22, “username”: “user12”, “email”: “user12@gmail.com”, “first_name”: “User”, “last_name”: “One”} به تبادل کاربر (user_exchange) ارسال می‌کند.

2.سرویس ایمیل (EmailService) باید از کلید پیوند “user.created.key” برای دریافت این پیام استفاده کند. پس از دریافت پیام، باید یک ایمیل خوش‌آمدگویی به کاربر ارسال کند.

3.سرویس لاگینگ (LoggingService) نیز باید از کلید پیوند “user.created.key” برای دریافت این پیام استفاده کند. پس از دریافت پیام، باید ورودی کاربر را در لاگر ثبت کند.

ما از Pika به عنوان مشتری Python برای RabbitMQ استفاده می‌کنیم. حتماً اطمینان حاصل کنید که pika، django و mysqlclient نصب شده باشند.

pip3 install django
pip3 install pika
pip3 install djangorestframework

بر اساس سیستم‌عامل خود، RabbitMQ را طبق توضیحات این صفحه نصب کنید.

ما یک گوش‌کننده پیام (Subscriber) را به عنوان یک نخ Python ایجاد خواهیم کرد، زیرا می‌خواهیم آن را به عنوان یک واحد جداگانه اجرا کنیم. این گوش‌کننده پیام به عنوان یک دستور سفارشی Django از ترمینال به شکل زیر اجرا خواهد شد:

python3 manage.py launch_queue_listener

منطق گوش‌کننده با استفاده از توضیحات مناسب توضیح داده شده است. برای کسب اطلاعات بیشتر، لطفاً این آموزش و آموزش‌های آینده را دنبال کنید.

mport json
import pika
import threading
ROUTING_KEY = 'user.created.key'
EXCHANGE = 'user_exchange'
THREADS = 5

class UserCreatedListener(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        self.channel = connection.channel()
        self.channel.exchange_declare(exchange=EXCHANGE, exchange_type='direct')
       # self.channel.queue_declare(queue=QUEUE_NAME, auto_delete=False)
        result = self.channel.queue_declare(queue='', exclusive=True)
        queue_name = result.method.queue
        self.channel.queue_bind(queue=queue_name, exchange=EXCHANGE, routing_key=ROUTING_KEY)
        self.channel.basic_qos(prefetch_count=THREADS*10)
        self.channel.basic_consume(queue=queue_name, on_message_callback=self.callback)
        
    def callback(self, channel, method, properties, body):
        #print(properties.content_type)
        #print(method)
        if(properties.content_type=="user_created_method"):
            message = json.loads(body)
            print(message)
        channel.basic_ack(delivery_tag=method.delivery_tag)
        
    def run(self):
        print ('Inside LogginService:  Created Listener ')
        self.channel.start_consuming()Python RabbitMQ Pika Listener Thread

دستور برای راه‌اندازی گوش‌دهنده به شکل زیر است:

from django.core.management.base import BaseCommand
from logger.queue_listener import UserCreatedListener
class Command(BaseCommand):
    help = 'Launches Listener for user_created message : RaabitMQ'
    def handle(self, *args, **options):
        td = UserCreatedListener()
        td.start()
        self.stdout.write("Started Consumer Thread")
view rawlaunch_queue_listener.py hosted with ❤ by GitHub

حالا بیایید تولیدکننده (Publisher) را به شکل زیر ایجاد کنیم:

import json
import pika

ROUTING_KEY = 'user.created.key'
EXCHANGE = 'user_exchange'
THREADS = 5
class ProducerUserCreated:
    def __init__(self) -> None:        
        # hearbeat = 600 indicates that after 600 seconds 
        # the peer TCP connection should be considered unreachable 
        #  by RabbitMQ and client libraries
        #
        #blocked_connection_timeout=300 means after 300 seconds 
        # the peer TCP connection is interrupted and dropped.
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost', heartbeat=600, blocked_connection_timeout=300)
            )
        self.channel = self.connection.channel()

    # This method will be called inside view for sending RabbitMQ message
    # method here is same as properties.content_type in listener callback
    def publish(self,method, body):
        print('Inside UserService: Sending to RabbitMQ: ')
        print(body)
        properties = pika.BasicProperties(method)
        self.channel.basic_publish(
            exchange=EXCHANGE, routing_key=ROUTING_KEY, body=json.dumps(body), 
            properties=properties)

ارسال پیام توسط ناشر RabbitMQ Pika هنگامی که یک کاربر ثبت نام می‌شود به صورت زیر است

منطق نمایش برای ارسال پیام:

from django.contrib.auth.models import User
from .serializers import RegisterSerializer
from rest_framework import generics
from .producer_user_created import ProducerUserCreated
from rest_framework.response import Response
from rest_framework import status
import json
producerUserCreated=ProducerUserCreated()
class RegisterView(generics.CreateAPIView):
    queryset = User.objects.all()
    #permission_classes = (AllowAny,)
    serializer_class = RegisterSerializer
    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        producerUserCreated.publish("user_created_method",json.dumps(serializer.data))
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

ناشر RabbitMQ Pika برای ارسال پیام هنگامی که یک کاربر جدید ثبت‌نام می‌کند.

چگونه اجرا کنیم؟

یک ترمینال اول را باز کنید تا RabbitMQ را راه‌اندازی کنید.

brew services start rabbitmq

ترمینال دوم را برای راه اندازی EmailService باز کنید

cd /path/to/EmailService
python3 manage.py launch_queue_listener

قلم سوم ترمینال برای راه اندازی LogginService

cd /path/to/LoggingService
python3 manage.py launch_queue_listener

ترمینال چهارم را باز کنید و UserService API را به صورت زیر راه اندازی کنید

cd /path/to/UserService
python3 manage.py runserver

حالا پستمن را باز کنید و درخواست POST را به URL زیر ارسال کنید:

با استفاده از پارامترهایی همچون:

http://127.0.0.1:8000/auth/register/

username:johntravolta99
password:Pa55word!@
password2:Pa55word!@
email:johntravolta99@gmail.com
first_name:John
last_name:Travolta

منبع مطلب