我在 RabbitMQ 中有一個任務隊列,其中有多個生產者 (12) 和一個消費者,用于 webapp 中的繁重任務。當我運行消費者時,它會在出現此錯誤之前開始將一些消息出列:Traceback (most recent call last):File "jobs.py", line 42, in <module> jobs[job](config)File "/home/ec2-user/project/queue.py", line 100, in init_queuechannel.start_consuming()File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 1822, in start_consumingself.connection.process_data_events(time_limit=None)File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 749, in process_data_eventsself._flush_output(common_terminator)File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 477, in _flush_outputresult.reason_text)pika.exceptions.ConnectionClosed: (-1, "error(104, 'Connection reset by peer')")生產者代碼是:message = {'image_url': image_url, 'image_name': image_name, 'notes': notes}connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))channel = connection.channel()channel.queue_declare(queue='tasks_queue')channel.basic_publish(exchange='', routing_key=queue_name, body=json.dumps(message))connection.close()唯一的消費者代碼(那個是沖突的):def callback(self, ch, method, properties, body): """Callback when receive a message.""" message = json.loads(body) try: image = _get_image(message['image_url']) except: sys.stderr.write('Error getting image in note %s' % note['id']) # Crop image with PIL. Not so expensive box_path = _crop(image, message['image_name'], box) # API call. Long time function result = long_api_call(box_path) if result is None: sys.stderr.write('Error in note %s' % note['id']) return # update the db db.update_record(result)如您所見,消息有 3 個昂貴的函數。一項裁剪任務、一項 API 調用和一項數據庫更新。沒有API調用,que消費者運行流暢。
添加回答
舉報
0/150
提交
取消