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
new file mode 100644
index 0000000..cbd2dff
--- /dev/null
+++ b/rabbitmq/receive.py
@@ -0,0 +1,37 @@
+import functools
+import time
+
+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()
+
+# channel.queue_declare(queue='dev_task')
+
+
+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)))
+
+
+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)
+
+
+# auto_ack=False 代表消息不自动确认
+channel.basic_consume(on_message_callback=do_work, queue='dev_task', auto_ack=False)
+
+channel.start_consuming()
diff --git a/rabbitmq/send.py b/rabbitmq/send.py
new file mode 100644
index 0000000..aadfa2e
--- /dev/null
+++ b/rabbitmq/send.py
@@ -0,0 +1,21 @@
+import pika
+
+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'), properties=properties)
+
+print(f"[x] send '{msg}'")
+connection.close()