Version 1

Copy# rabbit_util.py

import json

from pika.adapters.blocking_connection import BlockingConnection, BlockingChannel
from pika.credentials import PlainCredentials
from pika.connection import ConnectionParameters
from pika.spec import Basic, BasicProperties


class BaseConnection:
    DEFAULT_USERNAME = 'guest'
    DEFAULT_PASSWORD = 'guest'

    def __init__(self, host, port, username=DEFAULT_USERNAME, password=DEFAULT_PASSWORD, virtual_host='/'):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.virtual_host = virtual_host
        self.credentials = PlainCredentials(self.username, self.password)
        self.conn = BlockingConnection(ConnectionParameters(
            host=self.host,
            port=self.port,
            virtual_host=self.virtual_host,
            credentials=self.credentials,
        ))
        self.channel = self.conn.channel()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, value, traceback):
        if self.channel and self.channel.is_open:
            self.channel.stop_consuming()
            self.channel.close()
        if self.conn and self.conn.is_open:
            self.conn.close()


class Consumer(BaseConnection):
    def __init__(self, host, port, username, password, virtual_host):
        super().__init__(host, port, username=username, password=password, virtual_host=virtual_host)

    def start(self, queue, exchange='', routing_key=None):
        self.channel.exchange_declare(exchange, durable=True)  # 声明持久化交换机，不存在则创建
        self.channel.queue_declare(queue, durable=True)  # 声明持久化队列，不存在则创建
        self.channel.queue_bind(queue, exchange=exchange, routing_key=routing_key)  # 交换机与队列进行绑定
        self.channel.basic_qos(prefetch_count=1)  # 每次消费数量， 1为最佳，视业务情况
        self.channel.basic_consume(queue, self.consume_callback)  # 消费队列与消费回调
        self.channel.start_consuming()  # 开启消费者服务

    def consume_callback(self, channel: BlockingChannel, method: Basic.Deliver, properties: BasicProperties,
                         body: bytes):
        print(f'[CurrentConsume]: {method.consumer_tag} [Body]: {body}')
        body = json.loads(body.decode('utf-8'))
        print(body, type(body))
        channel.basic_ack(delivery_tag=method.delivery_tag)

    def run(self, queue, exchange='', routing_key=None):
        print('Consumer Service Started')
        print(' [*] Waiting for messages. To exit press CTRL+C')
        try:
            self.start(queue, exchange=exchange, routing_key=routing_key)
        except Exception as e:
            print(e)


class Producer(BaseConnection):
    def __init__(self, host, port, username, password, virtual_host):
        super().__init__(host, port, username=username, password=password, virtual_host=virtual_host)

    def publish_message(self, message, exchange='', routing_key=None):
        body = json.dumps(message).encode('utf-8')
        self.channel.basic_publish(
            exchange=exchange,
            routing_key=routing_key,
            body=body,
            properties=BasicProperties(delivery_mode=2)
        )

# consumer.py
from rabbit import Consumer


if __name__ == '__main__':
    with Consumer('192.168.79.128', 5672, username='admin', password='admin', virtual_host='my_vhost') as consumer:
        try:
            consumer.run('queue_1', exchange='exchange_1', routing_key='r_k_1')
        except KeyboardInterrupt as e:
            print(e)

# producer.py
import time

from rabbit import Producer


if __name__ == '__main__':
    with Producer('192.168.79.128', 5672, username='admin', password='admin', virtual_host='my_vhost') as producer:
        for i in range(10):
            time.sleep(2)
            producer.publish_message({'number': i}, exchange='exchange_1', routing_key='r_k_1')

Version2

# rabbit.py
import json
import logging
import traceback
from pika.adapters.blocking_connection import BlockingConnection, BlockingChannel
from pika.credentials import PlainCredentials
from pika.connection import ConnectionParameters
from pika.spec import Basic, BasicProperties
from pika.exchange_type import ExchangeType
from typing import Callable


class RabbitMQClient:
    DEFAULT_USERNAME = 'guest'
    DEFAULT_PASSWORD = 'guest'

    DEAD_EXCHANGE_NAME = "dead_exchange"
    DEAD_QUEUE_NAME = "dead_queue"
    DEAD_ROUTING_KEY = "dead_key"

    RETRY_EXCHANGE_NAME = "retry_exchange"
    RETRY_QUEUE_NAME = "retry_queue"
    RETRY_ROUTING_KEY = "retry_key"

    def __init__(self, host, port, username=DEFAULT_USERNAME, password=DEFAULT_PASSWORD, virtual_host='/'):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.virtual_host = virtual_host
        self.credentials = PlainCredentials(self.username, self.password)
        self.connection = BlockingConnection(ConnectionParameters(
            host=self.host,
            port=self.port,
            virtual_host=self.virtual_host,
            credentials=self.credentials,
        ))
        self.channel = self.connection.channel()
        self.channel.confirm_delivery()
        logging.debug("connection established")

    def close_connection(self):
        self.connection.close()
        logging.debug("connection closed")

    def declare_exchange(self, exchange, exchange_type=ExchangeType.direct.value, is_durable=False):
        self.channel.exchange_declare(exchange=exchange,
                                      exchange_type=exchange_type,
                                      durable=is_durable)

    def declare_queue(self, queue, arguments=None, is_durable=False):
        self.channel.queue_declare(queue=queue, durable=is_durable, arguments=arguments)

    def declare_retry_queue(self):
        """
        创建异常交换器和队列，用于存放没有正常处理的消息。
        :return:
        """
        self.channel.exchange_declare(exchange=self.RETRY_EXCHANGE_NAME,
                                      exchange_type=ExchangeType.direct.value,
                                      durable=True)
        self.channel.queue_declare(queue=self.RETRY_QUEUE_NAME,
                                   durable=True)
        self.channel.queue_bind(self.RETRY_QUEUE_NAME, exchange=self.RETRY_EXCHANGE_NAME, routing_key=self.RETRY_ROUTING_KEY)

    def publish_common_message(self, msg, queue, exchange='', routing_key=None):
        """
        发送消息到指定的交换器
        :param queue: 队列
        :param routing_key: RoutingKey
        :param exchange: RabbitMQ交换器
        :param msg: 消息实体
        :return:
        """
        self.declare_common_stuff(exchange=exchange, queue=queue, routing_key=routing_key)
        if not isinstance(msg, bytes):
            msg = json.dumps(msg).encode('utf-8')
        self.channel.basic_publish(exchange=exchange,
                                   routing_key=routing_key,
                                   body=msg,
                                   properties=BasicProperties(
                                       delivery_mode=2,
                                       type=exchange
                                   ))
        self.close_connection()
        print("message send out to %s" % exchange)
        # logging.debug("message send out to %s" % exchange)

    def publish_delay_message(self, msg, exchange='', queue='', routing_key=None, ttl=1 * 1000):
        """
        发布延时消息
        :param routing_key:
        :param queue:
        :param exchange:
        :param msg: 消息实体
        :param ttl: 延时时间，毫秒
        :return:
        """
        self.declare_delay_stuff()
        # self.delete_old(exchange=exchange, queue=queue)  # 删除旧交换机和旧队列，不建议。可以用随机值做为交换机和队列名，然后消费后删除该交换机和队列
        arguments = {
            'x-dead-letter-exchange': self.DEAD_EXCHANGE_NAME,
            'x-dead-letter-routing-key': self.DEAD_ROUTING_KEY,
            # 'x-message-ttl': ttl,
        }
        self.declare_publish_stuff(exchange=exchange, queue=queue, routing_key=routing_key, arguments=arguments)
        if not isinstance(msg, bytes):
            msg = json.dumps(msg).encode('utf-8')
        self.channel.basic_publish(exchange=exchange,
                                   routing_key=routing_key,
                                   body=msg,
                                   properties=BasicProperties(
                                       delivery_mode=2,
                                       expiration=str(ttl)
                                   ))
        # self.close_connection()
        print("message send out to %s" % exchange)
        # logging.debug("message send out to %s" % self.DEAD_EXCHANGE_NAME)

    def start_common_consume(self, callback: Callable, exchange='', queue='#', routing_key=None):
        """
        启动消费者，开始消费RabbitMQ中的消息
        :return:
        """
        self.declare_common_stuff(exchange=exchange, queue=queue, routing_key=routing_key)
        self.start_consume(queue, callback)

    def delete_old(self, exchange, queue):
        self.channel.exchange_delete(exchange=exchange)
        self.channel.queue_delete(queue=queue)

    def start_delay_consume(self, callback: Callable):
        self.declare_delay_stuff()
        self.start_consume(self.DEAD_QUEUE_NAME, callback)

    def start_retry_consume(self, callback: Callable):
        self.declare_retry_queue()
        self.start_consume(self.RETRY_QUEUE_NAME, callback)

    def start_consume(self, queue, callback: Callable):
        self.channel.basic_qos(prefetch_count=1)
        try:
            self.channel.basic_consume(
                queue,  # 你要从那个队列里收消息
                callback,  # 如果收到消息，就调用callback函数来处理消息
            )
            self.channel.start_consuming()
        except KeyboardInterrupt:
            self.stop_consuming()

    def declare_delay_stuff(self):
        self.declare_exchange(self.DEAD_EXCHANGE_NAME, exchange_type=ExchangeType.topic.value, is_durable=True)
        self.declare_queue(self.DEAD_QUEUE_NAME, is_durable=True)
        self.channel.queue_bind(queue=self.DEAD_QUEUE_NAME, exchange=self.DEAD_EXCHANGE_NAME,
                                routing_key=self.DEAD_ROUTING_KEY)

    def declare_common_stuff(self, exchange='', queue='#', routing_key=None):
        self.declare_exchange(exchange, exchange_type=ExchangeType.topic.value, is_durable=True)
        self.declare_queue(queue, is_durable=True)
        self.channel.queue_bind(queue, exchange=exchange, routing_key=routing_key)

    def declare_publish_stuff(self, exchange='', queue='#', routing_key=None, arguments=None):
        self.declare_exchange(exchange, exchange_type=ExchangeType.topic.value, is_durable=True)
        self.declare_queue(queue, arguments=arguments, is_durable=True)
        self.channel.queue_bind(queue, exchange=exchange, routing_key=routing_key)

    def stop_consuming(self):
        self.channel.stop_consuming()
        self.close_connection()

    @staticmethod
    def message_handle_successfully(channel: BlockingChannel, method: Basic.Deliver):
        """
        如果消息处理正常完成，必须调用此方法，
        否则RabbitMQ会认为消息处理不成功，重新将消息放回待执行队列中
        :param channel: 回调函数的channel参数
        :param method: 回调函数的method参数
        :return:
        """
        channel.basic_ack(delivery_tag=method.delivery_tag)

    @staticmethod
    def message_handle_failed(channel: BlockingChannel, method: Basic.Deliver):
        """
        如果消息处理失败，应该调用此方法，会自动将消息放入异常队列
        :param channel: 回调函数的channel参数
        :param method: 回调函数的method参数
        :return:
        """
        channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)


def consume_callback(channel: BlockingChannel, method: Basic.Deliver, properties: BasicProperties, body: bytes):
    print(f'[CurrentConsume]: {method.consumer_tag} [Body]: {body}')
    try:
        body = json.loads(body.decode('utf-8'))
        print(body, type(body))
        # 如果处理成功，则调用此消息回复ack，表示消息成功处理完成。
        RabbitMQClient.message_handle_successfully(channel, method)
    except Exception as e:
        traceback.print_exc()
        RabbitMQClient.message_handle_failed(channel, method)

        
# consumer.py
from rabbit import RabbitMQClient, consume_callback


def main():
    print("start program")
    client = RabbitMQClient('192.168.79.128', 5672, username='admin', password='admin', virtual_host='my_vhost')
    # client.start_common_consume(consume_callback, exchange='test1', queue='test1', routing_key='test1')
    client.start_delay_consume(consume_callback)


if __name__ == '__main__':
    main()


    
# producer.py
from uuid import uuid4
from rabbit import RabbitMQClient

if __name__ == '__main__':
    print("start publish")
    client = RabbitMQClient('192.168.79.128', 5672, username='admin', password='admin', virtual_host='my_vhost')
    msg1 = {"key": "value"}
    client.publish_delay_message(msg1, exchange='ttl_exchange', queue='ttl_exchange', routing_key='ttl_test', ttl=3 * 1000)
    # client.publish_common_message('test1', 'test1', 'test1', 'test1')
    print("message send out")