django项目中Rabbitmq的使用，两个项目之间数据传输存储

Dear 丶 2023-02-23 05:24 4阅读 0赞

解决问题：  
1、开发时会使用django环境进行一些初始化操作，这些程序一般只执行几次，但是需要django中的环境变量。  
2、使用django运行阻塞监听的程序，比如Rabbitmq监听，放在主程序中就阻塞住了，需要另外开命令执行。

│   ├── MyRab					# app名称
    │   │   ├── admin.py
    │   │   ├── apps.py
    │   │   ├── __init__.py
    │   │   ├── management			# 这个就是创建的 management 文件夹
    │   │   │   ├── commands		# commands文件夹
    │   │   │   │   ├── init_mq.py	# python manage.py init_mq 执行这个文件的内容，初始化
    │   │   │   │   ├── __init__.py		
    │   │   │   │   ├── intelligentAduit_listener.py # python manage.py intelligentAduit_listener 执行这个文件的内容 
    │   │   ├── migrations
    │   │   ├── models.py
    │   │   ├── __pycache__
    │   │   ├── serializers.py
    │   │   ├── task.py
    │   │   ├── tests.py
    │   │   └── views.py
    ........

**init\_mq.py 文件的内容**

from django.core.management.base import BaseCommand
    
    class Command(BaseCommand):
        def handle(self, *args, **options):
            from MyRab.management.untils.Rabbitmqserver import RabbitmqClient
            print("init_mq", "开始初始化Rabbitmq队列")
            try:
                RabbitmqClient.connent()
                RabbitmqClient.channel.queue_declare(queue='send_result', durable=True)
                RabbitmqClient.channel.queue_declare(queue='backend', durable=True)
                RabbitmqClient.channel.queue_declare(queue='intelligent', durable=True)
                RabbitmqClient.channel.queue_declare(queue='resume', durable=True)
                print("init_mq", "初始化Rabbitmq队列成功")
            except Exception as e:
                print("init_mq",e,"队列初始化失败")

初始化

python manage.py init_mq

**向rabbitmq队列发送数据**

import pika
    import sys
    import time
    
    if __name__ != '__main__':  # 测试服务是否能启动时使用
        from django.conf import settings
    
    
    class RabbitmqServer(object):
        def __init__(self, username, password, serverip, port):
            self.username = username
            self.password = password
            self.serverip = serverip
            self.port = port
    
        def connent(self):
            user_pwd = pika.PlainCredentials(self.username, self.password)
            s_conn = pika.BlockingConnection(
                pika.ConnectionParameters(host=self.serverip, port=self.port, credentials=user_pwd))  # 创建连接
            self.channel = s_conn.channel()
    
        def productMessage(self, queuename, message):
            self.channel.queue_declare(queue=queuename, durable=True)
            self.channel.basic_publish(exchange='',
                                       routing_key=queuename,  # 写明将消息发送给队列queuename
                                       body=message,  # 要发送的消息
                                       properties=pika.BasicProperties(delivery_mode=2, )
                                       # 设置消息持久化，将要发送的消息的属性标记为2，表示该消息要持久化
                                       )
    
        def expense(self, queuename, func):
            """ :param queuename: 消息队列名称 :param func: 要回调的方法名 """
            self.channel.basic_qos(prefetch_count=1)
            self.channel.basic_consume(auto_ack=False,  # 设置为应答（也就是必须处理完了，才将数据从队列里面删除）
                                       on_message_callback=func,
                                       queue=queuename,
                                       )
            self.channel.start_consuming()
    
    
    def callback(ch, method, properties, body):
        print(" [消费者] Received %r" % body)
        time.sleep(1)
        print(" [消费者] Done")
        ch.basic_ack(delivery_tag=method.delivery_tag)  # 接收到消息后会给rabbitmq发送一个确认
    
    
    if __name__ != '__main__':
        username = settings.RABBITMQCONFIG.get("username")
        password = settings.RABBITMQCONFIG.get("password")
        severip = settings.RABBITMQCONFIG.get("host")
        port = settings.RABBITMQCONFIG.get("port")
        RabbitmqClient = RabbitmqServer(username, password, severip, port)
    
    if __name__ == '__main__':
        import json
    
        RabbitmqClient = RabbitmqServer("username", "password", "127.0.0.1", 5672)
        RabbitmqClient.connent()
        data = { "code": 3}
        RabbitmqClient.productMessage("test3", json.dumps(data))
        RabbitmqClient.expense("test3", callback)

**在django项目中，向rabbitmq队列发送数据**

class ClassViewSet(viewsets.ModelViewSet):
        queryset = MyClass.objects.all()
        serializer_class = MyClassSer
        permission_classes = (AllowAny,)
    
        def create(self, request, *args, **kwargs):
            data = request.data
            print(data)
            print(type(data))
            serializer = self.get_serializer(data=request.data)
            serializer.is_valid(raise_exception=True)
            self.perform_create(serializer)
            headers = self.get_success_headers(serializer.data)
            # 将数据发送到rabbitmq消息队列
            sendtoparse_one(data)
            return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
    
    # 班级消息队列发送数据
    def sendtoparse_one(data):
        data = json.dumps(data)
        RabbitmqClient.connent()
        RabbitmqClient.productMessage("intelligent", data)
        # 发送消息通知算法端解析
        # logger.info("AduitOneParseView", "文件发送完成")
        print("AduitOneParseView", "文件发送完成")

当往项目A中创建班级信息的时候，就会将消息发送到“intelligent”队列中去

**接下来是项目B中，从队列里取数据，存入数据库**

阻塞监听队列，创建intelligentAduit\_listener.py文件

from django.core.management.base import BaseCommand
    from YouRab.management.untils.Rabbitmqserver import RabbitmqClient
    from YouRab.views import save_student_info, save_class_info
    
    class Command(BaseCommand):
        def handle(self, *args, **options):
            # 监听算法端返回的消息，将结果存入数据库并更改数据库中文件状态
            # 添加班级
            def parse_result_func(ch, method, properties, body):
                # 逻辑程序
                res = save_class_info(body)
                ch.basic_ack(delivery_tag=method.delivery_tag)  # 接收到消息后会给rabbitmq发送一个确认
    
            print("parse_result_func", "开始监听班级信息")
            RabbitmqClient.connent()
            RabbitmqClient.expense("intelligent", parse_result_func)

**django中，将数据存入数据库**

# 存入班级信息，在app中views里
    def save_class_info(data):
        data = json.loads(data)
        try:
            MyClass.objects.create(**data)
            res = "成功"
        except Exception as e:
            res = "失败"
        return res

**启动阻塞监听intelligent队列**

python manage.py intelligentAduit_listener

只要队列里面有了班级的消息信息，项目B就会将班级信息创建到数据库