MQ cơ bản - phần 2

Producer :

Producer là thành phần trong hệ thống chịu trách nhiệm tạo ra các thông điệp và gửi chúng đến Message Queue để được xử lý. Nói cách khác, Producer là "nguồn gốc" của Message. Các Producer thường là các thành phần hoặc dịch vụ trong hệ thống chịu trách nhiệm sản xuất Message dựa trên các sự kiện hoặc yêu cầu từ người dùng hoặc các thành phần khác.

Ví dụ, trong một ứng dụng thương mại điện tử, một Producer có thể là một thành phần quản lý giỏ hàng, tạo ra các thông điệp khi người dùng thêm sản phẩm vào giỏ hàng. Mỗi thông điệp này chứa thông tin về sản phẩm được thêm vào giỏ hàng và được gửi đến Message Queue để được xử lý bởi các Consumer tiếp theo.

Consumer :

Consumer là thành phần trong hệ thống chịu trách nhiệm lấy các thông điệp từ Message Queue và xử lý chúng. Các Consumer là "điểm đích" của thông điệp, nơi mà thông điệp được sử dụng hoặc xử lý theo cách mong muốn. Các Consumer thường là các thành phần hoặc dịch vụ trong hệ thống có khả năng xử lý các thông điệp cụ thể hoặc thực hiện các tác vụ cần thiết.

Tiếp tục ví dụ về ứng dụng thương mại điện tử, một Consumer có thể là một thành phần xử lý đơn hàng, lấy thông điệp từ Message Queue để tạo ra đơn hàng từ các sản phẩm trong giỏ hàng của người dùng. Một Consumer khác có thể là một thành phần gửi email xác nhận đơn hàng đến khách hàng sau khi đơn hàng đã được xử lý.

Tương tác giữa Producer và Consumer:

Producer và Consumer là hai mảnh ghép quan trọng trong hệ thống sử dụng Message Queue. Producer tạo ra thông điệp và gửi chúng đến Message Queue, trong khi Consumer lấy thông điệp từ hàng đợi và xử lý chúng. Quá trình này là không đồng bộ, nghĩa là Producer có thể tiếp tục tạo ra thông điệp mà không cần phải chờ đợi Consumer xử lý. Điều này giúp cải thiện hiệu suất và khả năng mở rộng của hệ thống.

Tóm lại, Producer và Consumer là hai vai trò quan trọng trong việc truyền tải thông điệp trong hệ thống sử dụng Message Queue. Producer tạo ra thông điệp và gửi chúng vào hàng đợi, trong khi Consumer lấy thông điệp từ hàng đợi và xử lý chúng theo cách mong muốn. Điều này giúp giảm bớt sự phụ thuộc giữa các thành phần, cải thiện hiệu suất và khả năng mở rộng của hệ thống.

Dưới đây là một ví dụ đơn giản về cách sử dụng RabbitMQ (một hệ thống Message Queue phổ biến) trong Python để tạo Producer và Consumer.

Trước tiên, bạn cần cài đặt thư viện pika, một thư viện Python cho RabbitMQ:

pip install pika

Sau đó, bạn có thể thực hiện các bước sau:

1. Producer:

import pika

# Kết nối tới RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Tạo một hàng đợi tên "hello"
channel.queue_declare(queue='hello')

# Gửi thông điệp đến hàng đợi "hello"
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')

print(" [x] Sent 'Hello World!'")

# Đóng kết nối
connection.close()

2. Consumer:

import pika

# Hàm callback được gọi khi nhận được thông điệp
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

# Kết nối tới RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Tạo một hàng đợi tên "hello"
channel.queue_declare(queue='hello')

# Đăng ký hàm callback cho hàng đợi "hello"
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')

# Bắt đầu lắng nghe các thông điệp từ hàng đợi
channel.start_consuming()

Trong ví dụ trên, Producer tạo một hàng đợi tên "hello" và gửi một thông điệp chứa chuỗi "Hello World!" vào hàng đợi. Consumer sau đó đăng ký một hàm callback để xử lý các thông điệp được nhận từ hàng đợi "hello" và in chúng ra màn hình. Khi chạy Producer và Consumer cùng một lúc, bạn sẽ thấy Consumer nhận và in ra thông điệp "Hello World!" từ Producer.

Dưới đây là một ví dụ về cách tạo Producer và Consumer sử dụng thư viện queue trong Python để thực hiện giao tiếp thông qua một hàng đợi đơn giản:

1. Producer:


import time
import random
import threading

# Khởi tạo một hàng đợi
message_queue = queue.Queue()

# Hàm sản xuất thông điệp
def produce():
    while True:
        # Tạo một thông điệp ngẫu nhiên
        message = f"Message {random.randint(1, 100)}"
        # Đưa thông điệp vào hàng đợi
        message_queue.put(message)
        print(f"Produced: {message}")
        # Ngủ 1-3 giây trước khi tạo thông điệp tiếp theo
        time.sleep(random.uniform(1, 3))

# Khởi tạo luồng sản xuất
producer_thread = threading.Thread(target=produce)
producer_thread.start()

2. Consumer:

# Hàm Consumer
def consume():
    while True:
        # Lấy một thông điệp từ hàng đợi
        message = message_queue.get()
        print(f"Consumed: {message}")
        # Xử lý thông điệp
        # Ở đây, chúng ta chỉ in ra thông điệp, nhưng thực tế có thể thực hiện các xử lý khác
        # ...
        # Đánh dấu thông điệp là đã xử lý xong
        message_queue.task_done()

# Khởi tạo luồng tiêu thụ
consumer_thread = threading.Thread(target=consume)
consumer_thread.start()

Trong ví dụ trên, Producer tạo ra các thông điệp ngẫu nhiên và đưa chúng vào hàng đợi, còn Consumer lấy các thông điệp từ hàng đợi và xử lý chúng. Hai luồng (threads) được sử dụng để thực hiện hai chức năng này một cách đồng thời. Bạn có thể thấy các thông điệp được sản xuất và tiêu thụ liên tục trong quá trình chạy của chương trình.

Message Queue cơ bản - part 2