Giới thiệu

Mẫu thiết kế Singleton là một trong những mẫu thiết kế phổ biến nhất trong phát triển phần mềm. Nó đảm bảo rằng một lớp chỉ có một thể hiện duy nhất và cung cấp một điểm truy cập toàn cầu đến nó. Trong Python đồng bộ, điều này khá đơn giản. Nhưng khi bạn bước vào thế giới lập trình bất đồng bộ, mọi thứ trở nên phức tạp hơn.

Trong bài viết này, chúng ta sẽ bắt đầu với những điều cơ bản về mẫu Singleton, sau đó chuyển vào thế giới async/await, và cuối cùng khám phá các mẫu sử dụng nâng cao cùng các kịch bản thực tế trong backend. Chúng ta cũng sẽ chỉ ra những cạm bẫy thường gặp và cách tránh chúng.

Những điều cơ bản: Singleton là gì?

Singleton là một lớp mà chỉ có thể có một thể hiện trong suốt vòng đời của một ứng dụng. Trong Python, điều này thường được thực hiện bằng cách ghi đè phương thức __new__ hoặc sử dụng metaclasses.

Singleton cổ điển (Đồng bộ)

class Singleton:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

Mẫu này hoạt động tốt khi mọi thứ là đồng bộ. Nhưng điều gì xảy ra nếu bạn cần khởi tạo singleton một cách bất đồng bộ, như khi thực hiện kết nối cơ sở dữ liệu?

Bước vào Python Bất Đồng Bộ

Python bất đồng bộ cho phép bạn viết mã không chặn bằng cách sử dụng async và await. Nó rất phù hợp cho các tác vụ I/O-bound như gọi API, truy vấn cơ sở dữ liệu hoặc thao tác tệp. Nhưng có một điều cần lưu ý: bạn không thể await bên trong __new__ hoặc __init__, vì chúng không phải là các hàm async. Vậy làm thế nào để bạn tạo một Singleton nhận thức bất đồng bộ?

Mẫu Singleton Bất Đồng Bộ

Giả sử chúng ta muốn tạo một Singleton quản lý một kết nối cơ sở dữ liệu. Việc thiết lập kết nối là bất đồng bộ. Ví dụ, sử dụng một client PostgreSQL bất đồng bộ.

Ví dụ 1: Singleton Kết Nối Cơ Sở Dữ Liệu Bất Đồng Bộ

import asyncio

class AsyncDBConnector:
    _instance = None
    _lock = asyncio.Lock()

    def __init__(self):
        self.connection = None

    @classmethod
    async def get_instance(cls):
        if cls._instance is None:
            async with cls._lock:
                if cls._instance is None:
                    instance = cls()
                    await instance._connect()
                    cls._instance = instance
        return cls._instance

    async def _connect(self):
        # Giả lập kết nối DB bất đồng bộ
        await asyncio.sleep(1)
        self.connection = "Đã kết nối với DB"

Tại sao nó hoạt động:

• Chúng ta sử dụng một phương thức lớp get_instance để tạo thể hiện.
• Chúng ta bọc nó bằng một khóa để đảm bảo an toàn với nhiều luồng trong môi trường đa tác vụ.
• Phương thức _connect là bất đồng bộ và thực hiện việc thiết lập.

Cách sử dụng:

async def main():
    db = await AsyncDBConnector.get_instance()
    print(db.connection)

asyncio.run(main())

Ví dụ 2: Singleton Tải Cấu Hình

Hãy xây dựng một singleton tải cấu hình mà đọc bí mật từ một API bất đồng bộ bên ngoài như AWS Secrets Manager.

class ConfigManager:
    _instance = None
    _lock = asyncio.Lock()

    def __init__(self):
        self.config = {}

    @classmethod
    async def get_instance(cls):
        if cls._instance is None:
            async with cls._lock:
                if cls._instance is None:
                    instance = cls()
                    await instance._load_config()
                    cls._instance = instance
        return cls._instance

    async def _load_config(self):
        await asyncio.sleep(0.5)  # Giả lập yêu cầu mạng
        self.config = {
            "DATABASE_URL": "postgresql://user:pass@host/db",
            "API_KEY": "abc123xyz"
        }

# Cách sử dụng
async def main():
    config = await ConfigManager.get_instance()
    print(config.config["API_KEY"])

asyncio.run(main())

Sử dụng Singleton Bất Đồng Bộ trong FastAPI

FastAPI là một trong những framework web bất đồng bộ phổ biến nhất trong Python. Nó hoạt động tốt với I/O không chặn và tiêm phụ thuộc. Hãy xem cách bạn có thể tích hợp mẫu Singleton bất đồng bộ vào ứng dụng FastAPI.

Trường hợp sử dụng: Client Redis hoặc Kết Nối DB Chia Sẻ

Giả sử bạn có một kết nối Redis mà nên được khởi tạo một lần và tái sử dụng trên tất cả các endpoint.

from fastapi import FastAPI, Depends
import asyncio

class RedisClient:
    _instance = None
    _lock = asyncio.Lock()

    def __init__(self):
        self.connection = None

    @classmethod
    async def get_instance(cls):
        if cls._instance is None:
            async with cls._lock:
                if cls._instance is None:
                    instance = cls()
                    await instance._connect()
                    cls._instance = instance
        return cls._instance

    async def _connect(self):
        await asyncio.sleep(1)
        self.connection = "Kết nối Redis giả lập"

# Wrapper phụ thuộc
async def get_redis_client() -> RedisClient:
    return await RedisClient.get_instance()

# Ứng dụng FastAPI
app = FastAPI()

@app.get("/status")
async def status(redis: RedisClient = Depends(get_redis_client)):
    return {"status": "ok", "redis": redis.connection}

• get_redis_client() được khai báo là một phụ thuộc bất đồng bộ.
• Thể hiện Singleton được giải quyết theo nhu cầu và chia sẻ sau đó.
• Điều này rất phù hợp cho Redis, PostgreSQL và các client HTTP tùy chỉnh như aiohttp hoặc httpx.

Những Cạm Bẫy Thường Gặp

1. Cố gắng await trong __init__

Điều này sẽ thất bại:

class BadExample:
    def __init__(self):
        await self.setup()  # SyntaxError

Bạn không thể await bên trong __init__. Sử dụng một phương thức thiết lập bất đồng bộ riêng.

2. Không sử dụng khóa

Nếu nhiều coroutine gọi get_instance() cùng một lúc, bạn có thể kết thúc với nhiều thể hiện. Luôn bảo vệ việc khởi tạo bằng cách sử dụng asyncio.Lock() để đảm bảo an toàn cho luồng.

Một vài mẹo

Làm Singleton thân thiện với kiểm tra

Đôi khi, thật hữu ích khi đặt lại hoặc ghi đè Singleton trong quá trình kiểm tra.

@classmethod
def reset_instance(cls):
    cls._instance = None

Tránh trạng thái toàn cục

Ngay cả với Singletons, hãy cẩn thận không lạm dụng chúng như các biến toàn cục. Tốt nhất là tiêm chúng nơi cần thiết như là các đối số của constructor.

Khởi tạo lười biếng

Trì hoãn công việc nặng nề cho đến khi thật sự cần thiết. Điều này giữ cho thời gian khởi động thấp.

Kết luận

Singleton trong Python bất đồng bộ không khó, nhưng chúng đòi hỏi bạn phải suy nghĩ lại cách và khi nào đối tượng của bạn được tạo ra. Điều quan trọng là: di chuyển tất cả logic await ra khỏi __init__ và vào một phương thức bất đồng bộ, và đảm bảo bảo vệ việc tạo thể hiện bằng asyncio.Lock().

Khi được sử dụng đúng cách, một Singleton bất đồng bộ có thể là một công cụ tuyệt vời để quản lý các tài nguyên chia sẻ như kết nối DB, tải cấu hình và client API trong các ứng dụng backend hiện đại.

Câu hỏi thường gặp (FAQ)

1. Mẫu Singleton có thể được sử dụng trong các trường hợp nào?
Mẫu Singleton thường được sử dụng trong các trường hợp cần chia sẻ trạng thái hoặc tài nguyên như kết nối cơ sở dữ liệu, cấu hình, và client API.

2. Có cách nào để kiểm tra sự hoạt động của Singleton không?
Có, bạn có thể sử dụng phương thức reset_instance để đặt lại thể hiện Singleton trong quá trình kiểm tra.

3. Làm thế nào để đảm bảo an toàn khi sử dụng Singleton bất đồng bộ?
Sử dụng asyncio.Lock() để bảo vệ việc khởi tạo thể hiện và đảm bảo an toàn cho nhiều luồng.

Bài viết gốc có thể được xem tại đây.