From d53b9fd7aa35227fd37171b8ac9d53f545814fe7 Mon Sep 17 00:00:00 2001 From: javaSwing Date: Thu, 13 Jun 2024 00:43:16 +0800 Subject: [PATCH 1/2] feat: add rabbitmq receive --- rabbitmq/receive.py | 71 +++++++++++++++++++++++++++++++++++++++++++++ rabbitmq/send.py | 13 +++++++++ 2 files changed, 84 insertions(+) create mode 100644 rabbitmq/receive.py create mode 100644 rabbitmq/send.py diff --git a/rabbitmq/receive.py b/rabbitmq/receive.py new file mode 100644 index 0000000..b01d361 --- /dev/null +++ b/rabbitmq/receive.py @@ -0,0 +1,71 @@ +import functools +import logging +import pika +import threading +import time + +LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ' + '-35s %(lineno) -5d: %(message)s') +LOGGER = logging.getLogger(__name__) + +logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT) + +# https://github.com/pika/pika/blob/0.12.0/examples/basic_consumer_threaded.py + +def ack_message(channel, delivery_tag): + """Note that `channel` must be the same pika channel instance via which + the message being ACKed was retrieved (AMQP protocol constraint). + """ + if channel.is_open: + channel.basic_ack(delivery_tag) + else: + # Channel is already closed, so we can't ACK this message; + # log and/or do something that makes sense for your app in this case. + pass + + +def do_work(connection, channel, delivery_tag, body:bytes): + thread_id = threading.get_ident() + fmt1 = 'Thread id: {} Delivery tag: {} Message body: {}' + LOGGER.info(fmt1.format(thread_id, delivery_tag, body.decode())) + # Sleeping to simulate 10 seconds of work + time.sleep(20) + cb = functools.partial(ack_message, channel, delivery_tag) + connection.add_callback_threadsafe(cb) + +def on_message(channel, method_frame, header_frame, body, args): + (connection, threads) = args + delivery_tag = method_frame.delivery_tag + t = threading.Thread(target=do_work, args=(connection, channel, delivery_tag, body)) + t.start() + threads.append(t) + +credentials = pika.PlainCredentials('guest', 'guest') +# Note: sending a short heartbeat to prove that heartbeats are still +# sent even though the worker simulates long-running work +parameters = pika.ConnectionParameters('localhost', credentials=credentials, heartbeat=5) +connection = pika.BlockingConnection(parameters) + +channel = connection.channel() +# channel.exchange_declare(exchange="test_exchange", exchange_type="direct", passive=False, durable=True, auto_delete=False) +# channel.queue_declare(queue="hello") +# channel.queue_bind(queue="hello", exchange='', routing_key='hello') +# Note: prefetch is set to 1 here as an example only and to keep the number of threads created +# to a reasonable amount. In production you will want to test with different prefetch values +# to find which one provides the best performance and usability for your solution +channel.basic_qos(prefetch_count=1) + +threads = [] +on_message_callback = functools.partial(on_message, args=(connection, threads)) +channel.basic_consume(queue='hello', on_message_callback=on_message_callback) + +try: + channel.start_consuming() +except KeyboardInterrupt: + channel.stop_consuming() + +# Wait for all to complete +for thread in threads: + thread.join() + +connection.close() \ No newline at end of file diff --git a/rabbitmq/send.py b/rabbitmq/send.py new file mode 100644 index 0000000..ae87d11 --- /dev/null +++ b/rabbitmq/send.py @@ -0,0 +1,13 @@ +import pika + +connection = pika.BlockingConnection(pika.connection.ConnectionParameters('localhost')) +channel = connection.channel() + +channel.queue_declare(queue='hello') + +msg = '你好，我是来自hello队列的消息' + +channel.basic_publish(exchange='', routing_key='hello', body=msg.encode('utf-8')) + +print(f"[x] send '{msg}'") +connection.close() From 89897bf268fc41a3acc49424ed8c99b73d564f76 Mon Sep 17 00:00:00 2001 From: javaSwing Date: Sun, 28 Jul 2024 19:09:06 +0800 Subject: [PATCH 2/2] feat: add rabbitmq receive reconnection --- .idea/misc.xml | 2 +- .idea/python-learn.iml | 6 +- rabbitmq/AutoRecoveryConsumer.py | 42 ++++++++++++ rabbitmq/consumer.py | 111 +++++++++++++++++++++++++++++++ rabbitmq/receive.py | 84 +++++++---------------- rabbitmq/send.py | 12 +++- 6 files changed, 193 insertions(+), 64 deletions(-) create mode 100644 rabbitmq/AutoRecoveryConsumer.py create mode 100644 rabbitmq/consumer.py diff --git a/.idea/misc.xml b/.idea/misc.xml index d56657a..845faad 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -1,4 +1,4 @@ - + \ No newline at end of file diff --git a/.idea/python-learn.iml b/.idea/python-learn.iml index 5ed0139..7bcaaaa 100644 --- a/.idea/python-learn.iml +++ b/.idea/python-learn.iml @@ -1,8 +1,10 @@ - - + + + + diff --git a/rabbitmq/AutoRecoveryConsumer.py b/rabbitmq/AutoRecoveryConsumer.py new file mode 100644 index 0000000..3815665 --- /dev/null +++ b/rabbitmq/AutoRecoveryConsumer.py @@ -0,0 +1,42 @@ +import time + +from rabbitmq.consumer import Consumer + + +class AutoRecoveryConsumer(object): + + def __init__(self, amqp_url, queue): + self._amqp_url = amqp_url + self._queue = queue + self._consumer = Consumer(self._amqp_url, queue) + + def run(self): + while True: + try: + self._consumer.run() + except KeyboardInterrupt: + self._consumer.stop() + break + self.maybe_reconnect() + + def maybe_reconnect(self): + if self._consumer.should_reconnect: + self._consumer.stop() + time.sleep(1) + self._consumer = Consumer(self._amqp_url, self._queue) + + +def main(): + username = 'guest' + password = 'guest' + host = 'localhost' + port = 5672 + vhost = '' + queue = 'hello' + amqp_url = 'amqp://{}:{}@{}:{}/{}'.format(username, password, host, port, vhost) + consumer = AutoRecoveryConsumer(amqp_url, queue) + consumer.run() + + +if __name__ == '__main__': + main() diff --git a/rabbitmq/consumer.py b/rabbitmq/consumer.py new file mode 100644 index 0000000..3705365 --- /dev/null +++ b/rabbitmq/consumer.py @@ -0,0 +1,111 @@ + +# https://www.alibabacloud.com/help/zh/apsaramq-for-rabbitmq/use-cases/automatic-recovery-from-network-failures#0a88a08a3eubs +import logging + +import pika +from pika import BaseConnection +from pika.channel import Channel + +LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ' + '-35s %(lineno) -5d: %(message)s') +LOGGER = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) + + +class Consumer(object): + + def __init__(self, amp_url, queue): + self.should_reconnect = False + + self._connection: BaseConnection | None = None + self._channel: Channel | None = None + self._closing = False + self.url = amp_url + + self._queue = queue + + def connect(self): + """ + 创建 connection,并设置回调 + """ + return pika.SelectConnection( + parameters=pika.URLParameters(self.url), + on_open_callback=self.on_connection_open, + on_open_error_callback=self.on_connection_open_failed, + on_close_callback=self.on_channel_closed + ) + + def on_connection_open(self, _unused_connection): + """ + 连接成功的回调 + """ + self._connection.channel(on_open_callback=self.on_channel_open) + + def on_connection_open_failed(self, _unused_connection: BaseConnection, err): + """ + 创建连接错误的回调 + """ + LOGGER.error("Connection failed: %s", err) + self.reconnect() + + def reconnect(self): + """ + 重连，修改 should_reconnect为 True，并停止 io_loop + """ + self.should_reconnect = True, + self.stop() + + def on_channel_open(self, channel: Channel): + """ + 创建 channel 之后的回调 + """ + self._channel = channel + self._channel.add_on_close_callback(self.on_channel_closed) + self.start_consuming() + + def on_channel_closed(self, channel, reason): + """ + channel 关闭的回调 + """ + LOGGER.warning('Channel %i was closed: %s', channel, reason) + self.close_connection() + + def start_consuming(self): + """ + 开始消费 + """ + LOGGER.info('start consuming...') + self._channel.basic_consume(queue=self._queue, on_message_callback=self.on_message, auto_ack=False) + + def close_connection(self): + """ + 关闭连接 + """ + if self._connection.is_closing or self._connection.is_closed: + LOGGER.info('Connection is closing or already closed') + else: + LOGGER.info('Closing connection') + self._connection.close() + + def on_message(self, _unused_channel, basic_deliver, properties, body): + """ + 消费消息并上传 ack + """ + LOGGER.info('Received message %s', body.decode()) + self._channel.basic_ack(delivery_tag=basic_deliver.delivery_tag) + + def run(self): + """ + 创建 connection，并启动 io_loop + """ + self._connection = self.connect() + self._connection.ioloop.start() + + def stop(self): + """ + 停止 io_loop + """ + if not self._closing: + self._closing = True + self._connection.ioloop.stop() + LOGGER.info('Stopping ioloop') diff --git a/rabbitmq/receive.py b/rabbitmq/receive.py index b01d361..cbd2dff 100644 --- a/rabbitmq/receive.py +++ b/rabbitmq/receive.py @@ -1,71 +1,37 @@ import functools -import logging -import pika -import threading import time -LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ' - '-35s %(lineno) -5d: %(message)s') -LOGGER = logging.getLogger(__name__) - -logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT) - -# https://github.com/pika/pika/blob/0.12.0/examples/basic_consumer_threaded.py - -def ack_message(channel, delivery_tag): - """Note that `channel` must be the same pika channel instance via which - the message being ACKed was retrieved (AMQP protocol constraint). - """ - if channel.is_open: - channel.basic_ack(delivery_tag) - else: - # Channel is already closed, so we can't ACK this message; - # log and/or do something that makes sense for your app in this case. - pass +import pika +from pika import spec +from pika.adapters.blocking_connection import BlockingChannel +connection = pika.BlockingConnection(pika.connection.ConnectionParameters(host='1781833300178911.mq-amqp.cn-hangzhou-a.aliyuncs.com',port=5672, virtual_host='yibms', credentials=pika.credentials.PlainCredentials( + 'MjoxNzgxODMzMzAwMTc4OTExOkxUQUk1dEI5ZFptV1pQd2dwYzFMcXcxZA==', + 'NzBGQjgzMjgxQzQyMEQ3OThDNDE4QjgzRUY2RjlEODBGOTgwNDAyOToxNzEzOTI1ODg2MTc5' + ))) +channel = connection.channel() -def do_work(connection, channel, delivery_tag, body:bytes): - thread_id = threading.get_ident() - fmt1 = 'Thread id: {} Delivery tag: {} Message body: {}' - LOGGER.info(fmt1.format(thread_id, delivery_tag, body.decode())) - # Sleeping to simulate 10 seconds of work - time.sleep(20) - cb = functools.partial(ack_message, channel, delivery_tag) - connection.add_callback_threadsafe(cb) +# channel.queue_declare(queue='dev_task') -def on_message(channel, method_frame, header_frame, body, args): - (connection, threads) = args - delivery_tag = method_frame.delivery_tag - t = threading.Thread(target=do_work, args=(connection, channel, delivery_tag, body)) - t.start() - threads.append(t) -credentials = pika.PlainCredentials('guest', 'guest') -# Note: sending a short heartbeat to prove that heartbeats are still -# sent even though the worker simulates long-running work -parameters = pika.ConnectionParameters('localhost', credentials=credentials, heartbeat=5) -connection = pika.BlockingConnection(parameters) +def callback(ch: BlockingChannel, method: spec.Basic.Deliver, properties: spec.BasicProperties, body: bytes): + connection.add_callback_threadsafe(functools.partial(do_work, (ch, method, properties, body))) -channel = connection.channel() -# channel.exchange_declare(exchange="test_exchange", exchange_type="direct", passive=False, durable=True, auto_delete=False) -# channel.queue_declare(queue="hello") -# channel.queue_bind(queue="hello", exchange='', routing_key='hello') -# Note: prefetch is set to 1 here as an example only and to keep the number of threads created -# to a reasonable amount. In production you will want to test with different prefetch values -# to find which one provides the best performance and usability for your solution -channel.basic_qos(prefetch_count=1) -threads = [] -on_message_callback = functools.partial(on_message, args=(connection, threads)) -channel.basic_consume(queue='hello', on_message_callback=on_message_callback) +def do_work(ch: BlockingChannel, method: spec.Basic.Deliver, properties: spec.BasicProperties, body: bytes): + try: + print(f'[x] received {body.decode()}') + time.sleep(20) + ch.basic_ack(delivery_tag=method.delivery_tag) + print(f'[ack] received {body.decode()}') + except Exception as e: + print(e) + finally: + print(ch.connection.is_open) + # ch.basic_ack(delivery_tag=method.delivery_tag) -try: - channel.start_consuming() -except KeyboardInterrupt: - channel.stop_consuming() -# Wait for all to complete -for thread in threads: - thread.join() +# auto_ack=False 代表消息不自动确认 +channel.basic_consume(on_message_callback=do_work, queue='dev_task', auto_ack=False) -connection.close() \ No newline at end of file +channel.start_consuming() diff --git a/rabbitmq/send.py b/rabbitmq/send.py index ae87d11..aadfa2e 100644 --- a/rabbitmq/send.py +++ b/rabbitmq/send.py @@ -1,13 +1,21 @@ import pika -connection = pika.BlockingConnection(pika.connection.ConnectionParameters('localhost')) +connection = pika.BlockingConnection(pika.connection.ConnectionParameters(host='localhost', + port=5672, + virtual_host='/', + credentials=pika.credentials.PlainCredentials( + 'guest', + 'guest' + ))) +properties = pika.BasicProperties(message_id='hellow') + channel = connection.channel() channel.queue_declare(queue='hello') msg = '你好，我是来自hello队列的消息' -channel.basic_publish(exchange='', routing_key='hello', body=msg.encode('utf-8')) +channel.basic_publish(exchange='', routing_key='hello', body=msg.encode('utf-8'), properties=properties) print(f"[x] send '{msg}'") connection.close()