Python/rq - theo dõi trạng thái của nhân viên

Question

Nếu đây là câu hỏi ngu ngốc, tôi xin lỗi và sẽ giấu đầu mình trong xấu hổ, nhưng:

Tôi đang sử dụng lệnh này để xếp hàng công việc bằng Python. Tôi muốn nó hoạt động như sau:

1. Công việc A bắt đầu. Job A lấy dữ liệu thông qua API web và lưu trữ nó.
2. Công việc A chạy.
3. Công việc A hoàn tất.
4. Sau khi hoàn thành A, lệnh B bắt đầu. Công việc B kiểm tra từng bản ghi được lưu trữ bởi công việc A và thêm một số dữ liệu phản hồi bổ sung.
5. Sau khi hoàn thành công việc B, người dùng nhận được một e-mail vui vẻ nói rằng báo cáo của họ đã sẵn sàng.

Mã của tôi cho đến nay:

redis_conn = Redis() 
use_connection(redis_conn) 
q = Queue('normal', connection=redis_conn) # this is terrible, I know - fixing later 
w = Worker(q) 
job = q.enqueue(getlinksmod.lsGet, theURL,total,domainid) 
w.work() 

tôi cho rằng giải pháp tốt nhất của tôi là phải có 2 công nhân, một cho công việc A và một cho B. Nhân viên công việc B có thể theo dõi công việc A và, khi công việc A đã được thực hiện, bắt đầu công việc B.

Những gì tôi không thể tìm ra để cứu mạng sống của mình là làm cách nào để một công nhân theo dõi tình trạng của người khác. Tôi có thể lấy ID công việc từ công việc A với job.id. Tôi có thể lấy tên công nhân với w.name. Nhưng không phải là người mù sương nhất về cách tôi chuyển bất kỳ thông tin nào cho người lao động khác.

Hoặc, có cách nào đơn giản hơn nhiều để làm điều này mà tôi hoàn toàn mất tích?

Answer 1

Bạn có thể quá sâu vào dự án của mình để chuyển đổi, nhưng nếu không, hãy xem Twisted. http://twistedmatrix.com/trac/ Tôi đang sử dụng nó ngay bây giờ cho dự án truy cập API, xóa nội dung web, v.v. Nó chạy nhiều công việc song song, cũng như tổ chức các công việc nhất định theo thứ tự, vì vậy Lệnh B không thực thi cho đến khi Lệnh A được thực hiện.

Đây là hướng dẫn tốt nhất cho việc học Twisted nếu bạn muốn thử. http://krondo.com/?page_id=1327

Answer 2

Kết hợp những thứ mà công việc A và công việc B thực hiện trong một chức năng và sau đó sử dụng ví dụ: multiprocessing.Pool (đó là phương pháp map_async) để trang trại ra ngoài các quy trình khác nhau.

Tôi không quen thuộc với rq, nhưng multiprocessing là một phần của thư viện chuẩn. Theo mặc định, nó sử dụng nhiều quá trình như CPU ​​của bạn có lõi, mà theo kinh nghiệm của tôi thường đủ để làm ướt máy.

Answer 3

Từ this page trên rq tài liệu, nó trông giống như từng đối tượng job có một thuộc tính result, callable bởi job.result, mà bạn có thể kiểm tra. Nếu công việc chưa hoàn thành, nó sẽ là None, nhưng nếu bạn đảm bảo rằng công việc của bạn trả về một số giá trị (thậm chí chỉ "Done"), thì bạn có thể nhờ nhân viên khác của bạn kiểm tra kết quả của công việc đầu tiên và sau đó bắt đầu làm việc chỉ khi job.result có giá trị, nghĩa là công nhân đầu tiên đã hoàn thành.

Answer 4

Cập nhật Januari 2015, yêu cầu kéo này hiện sáp nhập, và tham số được đổi tên thành depends_on, ví dụ:

second_job = q.enqueue(email_customer, depends_on=first_job) 

Các bài gốc còn nguyên vẹn cho những người chạy các phiên bản cũ và như vậy:

Tôi đã gửi yêu cầu kéo (https://github.com/nvie/rq/pull/207) để xử lý các phụ thuộc công việc trong RQ. Khi yêu cầu kéo này được hợp nhất, bạn sẽ có thể thực hiện:

def generate_report(): 
    pass 

def email_customer(): 
    pass 

first_job = q.enqueue(generate_report) 
second_job = q.enqueue(email_customer, after=first_job) 
# In the second enqueue call, job is created, 
# but only moved into queue after first_job finishes 

Hiện tại, tôi đề xuất viết hàm bao bọc để thực hiện công việc của bạn. Ví dụ:

def generate_report(): 
    pass 

def email_customer(): 
    pass 

def generate_report_and_email(): 
    generate_report() 
    email_customer() # You can also enqueue this function, if you really want to 

# Somewhere else 
q.enqueue(generate_report_and_email)