Trình ghi nhật ký không đồng bộ Python

Vì vậy, bạn đang xây dựng một ứng dụng dựa trên asyncio và giống như hầu hết các hệ thống chất lượng sản xuất, bạn cần ghi lại các sự kiện xuyên suốt. Thông thường, bạn sẽ tiếp cận mô-đun logging, nhưng mô-đun ghi nhật ký sử dụng I/O chặn khi gửi bản ghi

Hay không?

Khung này rất linh hoạt trong cách các bản ghi được phát ra; . Và kể từ Python 3. 2, một trình xử lý mới thú vị đã được đưa vào, lớp QueueHandler, đi kèm với lớp QueueListener tương ứng. Chúng ban đầu được phát triển để xử lý việc ghi nhật ký trong các quy trình con của thư viện multiprocessing, nhưng nếu không thì hoàn toàn có thể sử dụng được trong ngữ cảnh không đồng bộ

Lớp QueueListener bắt đầu chuỗi riêng của nó để xem hàng đợi và gửi bản ghi tới trình xử lý mà nó quản lý, vì vậy nó sẽ không ảnh hưởng đến vòng lặp asyncio. Việc triển khai trình xử lý QueueHandler chỉ đơn giản là đặt các bản ghi vào hàng đợi mà bạn chỉ định, sau một thao tác dọn dẹp tối thiểu để đảm bảo các bản ghi có thể được sắp xếp theo thứ tự một cách dễ dàng. Điều này làm cho trình xử lý này hoàn toàn không bị chặn

Di chuyển các trình xử lý hiện có sang QueueListener

Chiến lược của tôi chỉ đơn giản là di chuyển tất cả các trình xử lý gốc sang một đối tượng QueueListener trước khi vòng lặp asyncio chính bắt đầu và đặt một đối tượng QueueHandler vào vị trí của chúng. Từ đó trở đi, tất cả các hoạt động chặn trong trình xử lý được xử lý trong một chuỗi riêng biệt, giải phóng vòng lặp asyncio khỏi phải đợi các bản ghi nhật ký được ghi ra tệp hoặc ổ cắm mạng

Tôi có tùy chỉnh lớp QueueHandler một chút, nhưng chỉ ở mức tối thiểu thôi. không cần chuẩn bị các bản ghi đi vào hàng đợi cục bộ, đang xử lý, chúng tôi có thể bỏ qua quy trình đó và giảm thiểu chi phí ghi nhật ký hơn nữa

import asyncio
import logging
import logging.handlers
try:
    # Python 3.7 and newer, fast reentrant implementation
    # without task tracking (not needed for that when logging)
    from queue import SimpleQueue as Queue
except ImportError:
    from queue import Queue
from typing import List


class LocalQueueHandler(logging.handlers.QueueHandler):
    def emit(self, record: logging.LogRecord) -> None:
        # Removed the call to self.prepare(), handle task cancellation
        try:
            self.enqueue(record)
        except asyncio.CancelledError:
            raise
        except Exception:
            self.handleError(record)


def setup_logging_queue() -> None:
    """Move log handlers to a separate thread.

    Replace handlers on the root logger with a LocalQueueHandler,
    and start a logging.QueueListener holding the original
    handlers.

    """
    queue = Queue()
    root = logging.getLogger()

    handlers: List[logging.Handler] = []

    handler = LocalQueueHandler(queue)
    root.addHandler(handler)
    for h in root.handlers[:]:
        if h is not handler:
            root.removeHandler(h)
            handlers.append(h)

    listener = logging.handlers.QueueListener(
        queue, *handlers, respect_handler_level=True
    )
    listener.start()

Tất nhiên, bạn có thể định cấu hình mô-đun logging để bắt đầu sử dụng trình xử lý hàng đợi, nhưng tôi thấy mẫu trên hoạt động tốt hơn khi cũng sử dụng logging2 để xử lý cấu hình trình xử lý ở nơi khác. Sau đó, ở trên có thể chỉ cần lấy một hệ thống đã được định cấu hình và làm cho nó phù hợp với môi trường asyncio

Ghi nhật ký không đồng bộ là gì?

Python có phải là ghi nhật ký đơn lẻ không?

Ghi nhật ký getLogger (__ Tên __) Python là gì?

Ghi nhật ký Python có sử dụng log4j không?