From 735888e56fdd3fd7afbeecbbde9c8a78b41712bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Torma=20Krist=C3=B3f?= Date: Sat, 24 Oct 2020 00:29:47 +0200 Subject: [PATCH 1/3] use direct exchange --- cnn_classification_service/main.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cnn_classification_service/main.py b/cnn_classification_service/main.py index 0789212..17c3490 100644 --- a/cnn_classification_service/main.py +++ b/cnn_classification_service/main.py @@ -55,12 +55,12 @@ def main(): logging.info("Connecting to MQ service...") connection = pika.BlockingConnection(pika.connection.URLParameters(os.environ['PIKA_URL'])) channel = connection.channel() - channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='fanout') + channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='direct') queue_declare_result = channel.queue_declare(queue='', exclusive=True) queue_name = queue_declare_result.method.queue - channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'], queue=queue_name) + channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'],routing_key='feature', queue=queue_name) channel.basic_consume(queue=queue_name, on_message_callback=message_callback, auto_ack=True) From b20fbea277840f782d89a212286e91ea9ca0e030 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Torma=20Krist=C3=B3f?= Date: Sat, 24 Oct 2020 00:50:55 +0200 Subject: [PATCH 2/3] better load balancing --- cnn_classification_service/main.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/cnn_classification_service/main.py b/cnn_classification_service/main.py index 17c3490..5dd700d 100644 --- a/cnn_classification_service/main.py +++ b/cnn_classification_service/main.py @@ -27,6 +27,7 @@ def message_callback(channel, method, properties, body): routing_key='classification-result', body=json.dumps(results).encode("utf-8") ) + channel.basic_ack(delivery_tag = method.delivery_tag) def main(): @@ -57,12 +58,14 @@ def main(): channel = connection.channel() channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='direct') - queue_declare_result = channel.queue_declare(queue='', exclusive=True) + queue_declare_result = channel.queue_declare(queue='cnnqueue', exclusive=True) queue_name = queue_declare_result.method.queue channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'],routing_key='feature', queue=queue_name) - channel.basic_consume(queue=queue_name, on_message_callback=message_callback, auto_ack=True) + channel.basic_qos(prefetch_count=1) + + channel.basic_consume(queue=queue_name, on_message_callback=message_callback, auto_ack=False) logging.info("Connection complete! Listening to messages...") try: From 90c87038b6e5480b299151b4504b91f56ff2b961 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Torma=20Krist=C3=B3f?= Date: Sat, 24 Oct 2020 00:58:28 +0200 Subject: [PATCH 3/3] do not use exclusive queue --- cnn_classification_service/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cnn_classification_service/main.py b/cnn_classification_service/main.py index 5dd700d..f12f422 100644 --- a/cnn_classification_service/main.py +++ b/cnn_classification_service/main.py @@ -58,7 +58,7 @@ def main(): channel = connection.channel() channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='direct') - queue_declare_result = channel.queue_declare(queue='cnnqueue', exclusive=True) + queue_declare_result = channel.queue_declare(queue='cnnqueue', exclusive=False) queue_name = queue_declare_result.method.queue channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'],routing_key='feature', queue=queue_name)