Làm cách nào để tắt các quy trình công nhân Python RQ một cách chính xác?

Question

Sử dụng Python RQ, chúng tôi đang cố gắng quản lý động các quy trình của nhân viên. Chúng tôi sử dụng tập lệnh công nhân được xây dựng tùy chỉnh, trong đó (dưới dạng đơn giản) như sau:

from rq import Connection, Worker 

queues_to_listen_on = get_queues_to_listen_on() 

with Connection(connection = get_worker_connection()): 
    w = Worker(queues_to_listen_on) 
    w.work() 

Chúng tôi đặc biệt quan tâm đến việc tắt công nhân. Mối quan tâm chính của chúng tôi là làm thế nào để làm một tắt máy duyên dáng của một công nhân, theo cách mà sẽ cho phép công việc hiện tại được hoàn thành trước khi tắt. Handler request_stop(...) tín hiệu trên đối tượng thích hợp Worker dường như làm những gì chúng ta cần, nhưng có vẻ là không có cách nào (ít nhất là để hiểu biết của tôi) phát ra nó trừ khi nó là thông qua cách nhấn CTRL+C về quá trình lao động chạy trong nhà ga.

Như tôi đã nhìn thấy nó, có hai giải pháp có thể xảy ra (có chắc chắn có thể là nhiều hơn) - theo thứ tự ưu tiên:

1. lập trình, sử dụng thư viện rq, gửi tín hiệu để request_stop và do đó kích hoạt tắt máy duyên dáng .
2. Bằng cách nào đó có được pid của quá trình chính xác (không chắc chắn nếu quá trình Workhorse hoặc quá trình nghe người lao động) và sử dụng một số phương pháp khác, gửi tín hiệu thích hợp để quá trình đó. Chúng tôi có một số cách có thể được thực hiện, nhưng rất có thể sẽ yêu cầu nhiều công việc hơn và giới thiệu các biến khác cho vấn đề mà tôi muốn bị bỏ sót (ví dụ: sử dụng Fabric để chạy lệnh từ xa hoặc điều gì đó dọc theo các dòng đó).

Nếu có cách nào tốt hơn để giải quyết vấn đề này hoặc giải pháp thay thế khác sẽ đạt được mục tiêu tương tự, tôi sẽ đánh giá cao đề xuất của bạn.

Answer 1

Lựa chọn 1 là chắc chắn tốt hơn về mặt thiết kế.

Tuy nhiên để giải quyết vấn đề cụ thể của bạn vì phải sử dụng CTRL + C để thoát khỏi quá trình (Tôi ghét điều đó quá), bạn có thể sử dụng các chiến lược sau cho công nhân của bạn:

# WORKER_NAME.py 
import os 

PID = os.getpid() 

@atexit.register 
def clean_shut(): 
    print "Clean shut performed" 

    try: 
     os.unlink("WORKER_NAME.%d" % PID) 
    except: 
     pass 

# Worker main 
def main(): 
    f = open("WORKER_NAME.%d" % PID, "w") 
    f.write("Delete this to end WORKER_NAME gracefully") 
    f.close() 

    while os.path.exists("WORKER_NAME.%d" % PID): 
     # Worker working 

Và trong kịch bản tổng thể của bạn, nhận công nhân PID như @Borys được đề xuất, gửi yêu cầu dừng ấm và os.unlink("path/to/WORKER_NAME.%d" % worker_PID) để đảm bảo tắt máy duyên dáng :)

Điều này sẽ chỉ áp dụng cho công nhân chạy vòng lặp vô thời hạn. Nếu quy trình công nhân gọi những thứ chặn ngay cả công việc một lần liên tục, bạn phải theo dõi sâu hơn về thói quen chặn có thể để giải quyết từ đó, chẳng hạn như áp dụng một số loại chiến lược hết thời gian.