competitor_product_analysis/sdk/RabbitMQ.py

# -*- coding: utf-8 -*-
# @Time : 2022/12/2 15:03
# @Author : XuJiakai
# @File : RabbitMQ
# @Software: PyCharm

import pika
import json
import os

from pika.exchange_type import ExchangeType
from log import get_log

log = get_log('RabbitMQ')


def _default_callback(ch, method, properties, body):
    print('callback body ', body.decode())  # 在这里对消息内容进行我们想做的处理
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 手动应答ack，确保消息真正消费后才应答
    pass


class RabbitMQ(object):
    def __init__(self, username, password, host, port, virtual_host):
        self.username = username
        self.password = password
        self.host = host
        self.port = port
        self.virtual_host = virtual_host
        self.get_connect()  # 初始化RabbitMQ实例对象时完成连接rabbitmq服务器

    def get_connect(self):
        credentials = pika.PlainCredentials(username=self.username, password=self.password)  # 登录凭证
        self.connect = pika.BlockingConnection(
            pika.ConnectionParameters(host=self.host, port=self.port, virtual_host=self.virtual_host,
                                      credentials=credentials))
        self.channel = self.connect.channel()  # 客户端连接rabbitmq服务端后开辟管道，每个channel代表一个会话任务

    def send_by_fanout(self, exchange, body):
        """
        多消费、重复订阅方式
        :param exchange: 发送的topic
        :return:
        """
        self.channel.exchange_declare(
            exchange=exchange,
            exchange_type=ExchangeType.fanout.value,
        )
        # 声明队列， 并且队列持久化
        # self.channel.queue_declare(queue=queue, durable=False, auto_delete=True)
        # 通过routing_key 绑定 消息交换机和队列
        # self.channel.queue_bind(queue, exchange, routing_key)
        # 发消息
        self.channel.basic_publish(
            exchange=exchange,
            routing_key='',
            body=body,  # 发送的数据
        )
        pass

    def consumer_by_fanout(self, exchange, callback=_default_callback):
        """
        消费、扇出模式
        :param callback:
        :param exchange:
        :return:
        """
        self.channel.exchange_declare(exchange=exchange, exchange_type=ExchangeType.fanout.value)
        result = self.channel.queue_declare(queue='', auto_delete=True)
        queue_name = result.method.queue
        self.channel.queue_bind(queue_name, exchange, '')

        log.info('消费任务启动，queue: ' + queue_name)

        self.channel.basic_consume(
            queue=queue_name,  # 队列名
            on_message_callback=callback,  # 指定回调函数
            auto_ack=False,  # 关闭自动ack采用手动应答
        )
        self.channel.start_consuming()

    pass


import threading

if __name__ == '__main__':
    mq = RabbitMQ(username='whc', password='whc', host='106.15.78.184', port=5672, virtual_host='/')

    # p = threading.Thread(target=mq.consumer, args=('xjk_test', '', callback,))
    # p.start()
    mq.consumer_by_fanout(exchange='xjk_test')

    # mq.send(exchange='xjk_test', body='xjk_test message')

    pass