Làm thế nào để xử lý các kết nối SQLAlchemy trong ProcessPool?

Question

Tôi có một lò phản ứng mà lấy về các thông điệp từ một nhà môi giới RabbitMQ và gây nên các phương pháp nhân để xử lý các thông điệp trong một hồ bơi quá trình, một cái gì đó như thế này:

này được thực hiện sử dụng python asyncio, loop.run_in_executor() và concurrent.futures.ProcessPoolExecutor.

Bây giờ tôi muốn truy cập cơ sở dữ liệu trong các phương thức công nhân sử dụng SQLAlchemy. Chủ yếu là xử lý sẽ rất đơn giản và nhanh chóng hoạt động CRUD.

Lò phản ứng sẽ xử lý 10-50 tin nhắn mỗi giây ngay từ đầu, do đó không thể mở kết nối cơ sở dữ liệu mới cho mọi yêu cầu. Thay vào đó, tôi muốn duy trì một kết nối liên tục cho mỗi quá trình.

Câu hỏi của tôi là: Làm thế nào tôi có thể thực hiện việc này? Tôi có thể lưu trữ chúng trong một biến toàn cục không? Hồ bơi kết nối SQA có xử lý điều này cho tôi không? Làm thế nào để làm sạch khi lò phản ứng dừng lại?

[Cập nhật]

• Cơ sở dữ liệu là MySQL với InnoDB.

Tại sao nên chọn mẫu này với bể xử lý?

Triển khai hiện tại sử dụng một mẫu khác nhau nơi mỗi người tiêu dùng chạy theo chuỗi riêng của mình. Bằng cách nào đó điều này không hoạt động rất tốt. Đã có khoảng 200 người tiêu dùng mỗi người chạy theo chủ đề của riêng họ và hệ thống đang phát triển nhanh chóng. Để mở rộng quy mô tốt hơn, ý tưởng là phân tách các mối quan tâm và tiêu thụ thông báo trong vòng lặp I/O và ủy quyền xử lý cho một nhóm. Tất nhiên, hiệu suất của toàn bộ hệ thống chủ yếu là I/O bị ràng buộc. Tuy nhiên, CPU là một vấn đề khi xử lý các tập kết quả lớn.

Lý do khác là "dễ sử dụng". Trong khi việc xử lý kết nối và tiêu thụ thông điệp được triển khai không đồng bộ, mã trong công nhân có thể đồng bộ và đơn giản.

Chẳng bao lâu nó trở nên hiển nhiên khi truy cập hệ thống từ xa thông qua kết nối mạng liên tục từ bên trong công nhân là một vấn đề. Đây là những gì các CommunicationChannels là cho: Bên trong công nhân, tôi có thể cấp yêu cầu cho các tin nhắn xe buýt thông qua các kênh này.

Một trong những ý tưởng hiện tại của tôi là xử lý truy cập DB theo cách tương tự: Chuyển câu lệnh qua hàng đợi đến vòng lặp sự kiện nơi chúng được gửi tới DB. Tuy nhiên, tôi không có ý tưởng làm thế nào để làm điều này với SQLAlchemy. Đâu là điểm vào? Đối tượng cần phải là pickled khi chúng được chuyển qua hàng đợi. Làm cách nào để có được một đối tượng như vậy từ một truy vấn SQA? Giao tiếp với cơ sở dữ liệu phải hoạt động không đồng bộ để không chặn vòng lặp sự kiện. Tôi có thể sử dụng ví dụ: aiomysql như một trình điều khiển cơ sở dữ liệu cho SQA?

Answer 1

yêu cầu của bạn của kết nối một cơ sở dữ liệu cho mỗi quá trình quá trình bơi có thể dễ dàng thỏa mãn nếu một số chăm sóc được thực hiện vào cách bạn nhanh chóng session, giả sử bạn đang làm việc với orm, trong quá trình lao động.

Một giải pháp đơn giản sẽ có một thế giới session mà bạn sử dụng lại trên yêu cầu:

# db.py 
engine = create_engine("connection_uri", pool_size=1, max_overflow=0) 
DBSession = scoped_session(sessionmaker(bind=engine)) 

Và vào các nhiệm vụ công nhân:

# task.py 
from db import engine, DBSession 
def task(): 
    DBSession.begin() # each task will get its own transaction over the global connection 
    ... 
    DBSession.query(...) 
    ... 
    DBSession.close() # cleanup on task end 

Arguments pool_size và max_overflowcustomize mặc định QueuePool sử dụng bởi create_engine. pool_size sẽ đảm bảo quá trình của bạn chỉ giữ được 1 kết nối còn tồn tại cho mỗi quá trình trong nhóm xử lý.

Nếu bạn muốn kết nối lại, bạn có thể sử dụng DBSession.remove() để xóa phiên khỏi sổ đăng ký và sẽ kết nối lại vào lần sử dụng DBSession tiếp theo. Bạn cũng có thể sử dụng đối số recycle của Pool để kết nối lại kết nối sau khoảng thời gian được chỉ định.

Trong quá trình phát triển/gỡ lỗi, bạn có thể sử dụng AssertionPool sẽ tăng ngoại lệ nếu nhiều kết nối được kiểm tra từ hồ bơi, xem switching pool implementations về cách thực hiện điều đó.

Answer 2

@roman: Rất vui khi bạn có mặt ở đó.

Tôi đã ở trong một tình huống tương tự trước vì vậy đây là 2 của tôi cent: trừ khi chỉ tiêu dùng này "đọc" và "viết" tin nhắn, mà không làm bất cứ proccessing thực sự của nó, bạn có thể tái thiết kế người tiêu dùng này là người tiêu dùng/nhà sản xuất sẽ tiêu thụ thông báo, nó sẽ xử lý thông báo và sau đó sẽ đưa kết quả vào hàng đợi khác, hàng đợi đó (tin nhắn được xử lý để nói) có thể được đọc bởi 1..N non các quy trình không đồng bộ chung có thể mở kết nối DB trong toàn bộ vòng đời của nó.

Tôi có thể mở rộng câu trả lời, nhưng tôi không biết liệu phương pháp này có phù hợp với nhu cầu của bạn hay không, nếu có, tôi có thể cung cấp cho bạn chi tiết hơn về thiết kế mở rộng.

Answer 3

Phương pháp tiếp cận đã phục vụ tôi thực sự tốt là sử dụng máy chủ web để xử lý và mở rộng quy mô nhóm xử lý. bình-sqlalchemy ngay cả trong trạng thái mặc định của nó sẽ giữ một hồ bơi kết nối và không đóng mỗi kết nối trên mỗi chu kỳ phản ứng yêu cầu.

Trình thực thi asyncio chỉ có thể gọi các điểm kết thúc url để thực thi các chức năng của bạn. Lợi ích bổ sung là vì tất cả các quy trình thực hiện công việc đều nằm sau url, bạn có thể chia tỷ lệ hồ bơi công nhân của mình trên các máy mutliple, thêm nhiều quy trình hơn thông qua gunicorn hoặc một trong nhiều phương pháp khác để mở rộng một máy chủ wsgi đơn giản. Thêm vào đó bạn nhận được tất cả lòng tốt khoan dung lỗi.

Nhược điểm là bạn có thể chuyển nhiều thông tin hơn trên mạng. Tuy nhiên khi bạn nói vấn đề là CPU bị ràng buộc và có thể bạn sẽ chuyển nhiều dữ liệu hơn đến và đi từ cơ sở dữ liệu.