RabbitMQ的简单使用 —— Python篇

（一）RabbitMQ的简介
RabbitMq 是实现了高级消息队列协议（AMQP）的开源消息代理中间件。消息队列是一种应用程序对应用程序的通行方式，应用程序通过写消息，将消息传递于队列，由另一应用程序读取 完成通信。而作为中间件的 RabbitMq 无疑是目前最流行的消息队列之一。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ。

RabbitMQ总体架构

PS：生产者和消费者可能在不同的程序或主机中，当然也有可能一个程序有可能既是生产者，也是消费者。 

RabbitMq 应用场景广泛：
1.系统的高可用：日常生活当中各种商城秒杀，高流量，高并发的场景。当服务器接收到如此大量请求处理业务时，有宕机的风险。某些业务可能极其复杂，但这部分不是高时效性，不需要立即反馈给用户，我们可以将这部分处理请求抛给队列，让程序后置去处理，减轻服务器在高并发场景下的压力。

2.分布式系统，集成系统，子系统之间的对接，以及架构设计中常常需要考虑消息队列的应用。（二）RabbitMQ的安装

apt-get update
apt-get install erlang
apt-get install rabbitmq-server


#启动rabbitmq： service rabbitmq-server start
#停止rabbitmq： service rabbitmq-server stop
#重启rabbitmq： service rabbitmq-server restart

#启动rabbitmq插件：rabbitmq-plugins enable rabbitmq_management

启用rabbitmq_management插件后就可以登录后台管理页面了，浏览器输入ip:15672

自带的密码和用户名都是guest，但是只能本机登录

所以下面我们添加新用户，和自定义权限

#添加新用户
rabbitmqctl add_user 用户名 密码

#给指定用户添加管理员权限
rabbitmqctl set_user_tags 用户名 administrator

给用户添加权限
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"

在web页面输入用户名，和密码

（三）python操作RabbitMQ
python中使用pika操作RabbitMQ

pip install pika
#皮卡皮卡，哈哈

（四）RabbitMQ简单模式

上代码

# coding=utf-8
### 生产者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')#用户名和密码
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))#连接服务器上的RabbitMQ服务

# 创建一个channel
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='hello')

for i in range(0, 100):
    channel.basic_publish(exchange='',#当前是一个简单模式，所以这里设置为空字符串就可以了
                          routing_key='hello',# 指定消息要发送到哪个queue
                          body='{}'.format(i)# 指定要发送的消息
                          )
    time.sleep(1)
    
# 关闭连接
# connection.close()

PS：RabbitMQ中所有的消息都要先通过交换机，空字符串表示使用默认的交换机
 

# coding=utf-8
### 消费者

import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='hello')


# 回调函数
def callback(ch, method, properties, body):
    print('消费者收到:{}'.format(body))

# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息


channel.basic_consume(queue='hello',  # 接收指定queue的消息
                      auto_ack=True,  # 指定为True，表示消息接收到后自动给消息发送方回复确认，已收到消息
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

print('Waiting for messages. To exit press CTRL+C')

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

对于上面的这种模式，有一下两个不好的地方： 

一个是在我们的消费者还没开始消费完队列里的消息，如果这时rabbitmq服务挂了，那么消息队列里的消息将会全部丢失，解决方法是在声明队列时，声明队列为可持久化存储队列，并且在生产者将消息插入到消息队列时，设置消息持久化存储，具体如下 

# coding=utf-8
### 生产者
import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

# 创建一个channel
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='durable_queue',durable=True)
#PS：这里不同种队列不允许名字相同


for i in range(0, 100):
    channel.basic_publish(exchange='',
                          routing_key='durable_queue',
                          body='{}'.format(i),
                          properties=pika.BasicProperties(delivery_mode=2)
                          )


# 关闭连接
# connection.close()

消费者与上面的消费者没有什么不同，具体的就是消费声明的队列，也要是可持久化的队列，还有就是，即使在生产者插入消息时，设置当前消息持久化存储（properties=pika.BasicProperties(delivery_mode=2)），并不能百分百保证消息真的被持久化，因为RabbitMQ挂掉的时候它可能还保存在缓存中，没来得及同步到磁盘中

在生产者插入消息后，立刻停止rabbitmq，并重新启动，其实我们在web管理页面也可看到未被消费的信息，当然在启动消费者后也成功接收到了消息


上面说的第二点不好就是，如果在消费者获取到队列里的消息后，在回调函数的处理过程中，消费者突然出错或程序崩溃等异常，那么就会造成这条消息并未被实际正常的处理掉。为了解决这个问题，我们只需在消费者basic_consume(auto_ack=False),并在回调函数中设置手动应答即可ch.basic_ack(delivery_tag=method.delivery_tag)，具体如下

# coding=utf-8
### 消费者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='queue')


# 回调函数
def callback(ch, method, properties, body):
    time.sleep(5)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消费者收到:{}'.format(body.decode('utf-8')))


# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息


channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定为False，表示取消自动应答，交由回调函数手动应答
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

# 应答的本质是告诉消息队列可以将这条消息销毁了

print('Waiting for messages. To exit press CTRL+C')

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

这里只需要配置消费者，生产者并不要修改

还有就是在上的使用方式在，都是一个生产者和一个消费者，还有一种情况就是，一个生产者和多个消费者，即多个消费者同时监听一个消息队列，这时候队列里的消息就是轮询分发（即如果消息队列里有100条信息，如果有2个消费者，那么每个就会收到50条信息），但是在某些情况下，不同的消费者处理任务的能力是不同的，这时还按照轮询的方式分发消息并不是很合理，那么只需要再配合手动应答的方式，设置消费者接收的消息没有处理完，队列就不要给我放送新的消息即可，具体配置方式如下：

# coding=utf-8
### 消费者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='queue')


# 回调函数
def callback(ch, method, properties, body):
    time.sleep(0)#通过设置休眠时间来模拟不同消费者的处理时间
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消费者收到:{}'.format(body.decode('utf-8')))


# prefetch_count表示接收的消息数量，当我接收的消息没有处理完（用basic_ack标记消息已处理完毕）之前不会再接收新的消息了
channel.basic_qos(prefetch_count=1)  # 还有就是这个设置必须在basic_consume之上，否则不生效

channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定为False，表示取消自动应答，交由回调函数手动应答
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

# 应答的本质是告诉消息队列可以将这条消息销毁了

print('Waiting for messages. To exit press CTRL+C')

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

PS:这种情况必须关闭自动应答ack，改成手动应答。使用basicQos(perfetch=1)限制每次只发送不超过1条消息到同一个消费者，消费者必须手动反馈告知队列，才会发送下一个

（五）RabbitMQ发布订阅模式
发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中

这个模式中会引入交换机的概念，其实在RabbitMQ中，所有的生产者都不会直接把消息发送到队列中，甚至生产者都不知道消息在发出后有没有发送到queue中，事实上，生产者只能将消息发送给交换机，由交换机来决定发送到哪个队列中。 

交换机的一端用来从生产者中接收消息，另一端用来发送消息到队列，交换机的类型规定了怎么处理接收到的消息，发布订阅模式使用到的交换机类型为 fanout ，这种交换机类型非常简单，就是将接收到的消息广播给已知的（即绑定到此交换机的）所有消费者。

当然，如果不想使用特定的交换机，可以使用 exchange=‘’ 表示使用默认的交换机，默认的交换机会将消息发送到 routing_key 指定的queue，可以参考简单模式。

上代码：

#生产者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 创建一个指定名称的交换机，并指定类型为fanout，用于将接收到的消息广播到所有queue中
channel.exchange_declare(exchange='交换机', exchange_type='fanout')


# 将消息发送给指定的交换机，在fanout类型中，routing_key=''表示不用发送到指定queue中，
# 而是将发送到绑定到此交换机的所有queue
channel.basic_publish(exchange='交换机', routing_key='', body='这是一条测试消息')
#消费者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

channel.exchange_declare(exchange='交换机', exchange_type='fanout')

# 使用RabbitMQ给自己生成一个专有的queue
result = channel.queue_declare(queue='333')
# result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
# 这里如果设置exclusive=True参数，那么该队列就是一个只有队列，在消费者结束后，该专有队列也会自动清除，如果queue=''没有设置名字的话，那么就会自动生成一个
# 不会重复的队列名

# 将queue绑定到指定交换机
channel.queue_bind(exchange='交换机', queue=queue_name)

print(' [*] Waiting for  message.')


def callback(ch, method, properties, body):
    print("消费者收到:{}".format(body.decode('utf-8')))


channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

该模式与简单模式的还有一个区别就是，这里的消息队列都是由消费者声明的，所以如果是生产者先启动，并将消息发给交换机的画，这里的消息就会丢失，所以我们也可以在消费者端声明队列并绑定交换机（不能是专有队列），所以仔细想想，其实这所谓的发布订阅模式并没有说什么了不起，它不过是让交换机同时推送多条消息给绑定的队列，我们当然也可以在简单模式的基础上多进行几次basic_publish发送消息到指定的队列。当然我们这样做的话，可能就没办法做到由交换机的同时发送了，效率可能也没有一次basic_publish的高

（六）RabbitMQ RPC模式

下面实现由rpc远程调用加减运算 

客户端

import pika
import uuid
import json


class RPC(object):

    def __init__(self):
        self.call_id = None
        self.response = None
        user_info = pika.PlainCredentials('root', 'root')
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
        self.channel = self.connection.channel()

        # 创建一个此客户端专用的queue，用于接收服务端发过来的消息
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True)

    def on_response(self, ch, method, props, body):
        # 判断接收到的response是否属于对应request
        if self.call_id == props.correlation_id:
            self.response = json.loads(body.decode('utf-8')).get('result')

    def call(self, func, param):
        self.response = None
        self.call_id = str(uuid.uuid4())  # 为该消息指定uuid，类似于请求id
        self.channel.queue_declare(queue='rpc_queue')
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',  # 将消息发送到该queue
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,  # 从该queue中取消息
                correlation_id=self.call_id,  # 为此次消息指定uuid
            ),
            body=json.dumps(
                {
                    'func': func,
                    'param': {'a': param[0], 'b': param[1]}
                }
            )
        )
        
        self.connection.process_data_events(time_limit=3)# 与start_consuming()相似，可以设置超时参数
        return self.response


rpc = RPC()

print("发送消息到消费者,等待返回结果")

response = rpc.call(func='del', param=(1, 2))

print("收到来自消费者返回的结果:{}".format(response))


服务端

import pika
import json

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

channel = connection.channel()

# 指定接收消息的queue
channel.queue_declare(queue='rpc_queue')


def add_number(a, b):
    return a + b


def del_num(a, b):
    return a - b


execute_map = {
    'add': add_number,
    'del': del_num
}


def on_request(ch, method, props, body):
    body = json.loads(body.decode('utf-8'))
    func = body.get('func')
    param = body.get('param')
    result = execute_map.get(func)(param.get('a'), param.get('b'))
    print('进行{}运算，并将结果返回个消费者'.format(func))

    ch.basic_publish(exchange='',  # 使用默认交换机
                     routing_key=props.reply_to,  # response发送到该queue
                     properties=pika.BasicProperties(
                         correlation_id=props.correlation_id),  # 使用correlation_id让此response与请求消息对应起来
                     body=json.dumps({'result': result}))

    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)
# 从rpc_queue中取消息，然后使用on_request进行处理
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print(" [x] Awaiting RPC requests")
channel.start_consuming()

 本次分享到此结束，感谢大家的阅读！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/726595.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

nacos 配置修改.代码实时刷新

nacos 配置修改.代码实时刷新

再类上用 RefreshScope 更新Value(“${uniqlo.privacy.url:https://wsurl.cc/yourls-api.php}”) private String shortLinkGenerateUrl;的数据可以实时更新 2.再配置类上次用 ConfigurationProperties(prefix “test.privacy”) 和nacos的配置一直 Data Component Configu…
阅读更多...
redis高可用-哨兵机制

redis高可用-哨兵机制

一：背景 上一节我们已经实现了redis的主从同步，从而实现服务的流量分摊和数据高可用，但是出现故障以后，需要人工手动接入，手动切换主从，来实现故障转移。这是比较麻烦的，毕竟人不能实时盯着服务…
阅读更多...
深度学习算法informer（时序预测）（四）（Decoder）

深度学习算法informer（时序预测）（四）（Decoder）

一、DecoderLayer架构如图（不改变输入形状） 二、Decoder整体 包括两部分 1. 多层DecoderLayer 2. 层归一化 代码如下 class DecoderLayer(nn.Module):def __init__(self, self_attention, cross_attention, d_model, d_ffNone,dropout0.1, activati…
阅读更多...
java溯本求源之基础（二十五）之--ArrayList常用方法介绍

java溯本求源之基础（二十五）之--ArrayList常用方法介绍

1. 介绍 1.1简介 ArrayList ArrayList 是 Java 集合框架中的一个类，位于 java.util 包中。它实现了 List 接口，提供了一个动态数组的功能。与普通数组不同，ArrayList 可以在需要时自动调整其容量，以容纳更多的元素。这使得它非常…
阅读更多...
《模拟联合国2.9—团队协作》

《模拟联合国2.9—团队协作》

感谢上海财经大学持续的邀请，今天在阶梯教室举办的《模拟联合国2.0—团队协作》沙盘课程圆满结束。尽管场地的限制带来了一定的挑战，但得益于系统思考中“结构影响行为”的原则，我得以在不同场景中巧妙设计课程结构，极大地促进了大…
阅读更多...
【SEMI-e ·国际半导体深圳展】| 06月26-28日唯创知音语音芯片供应商 邀您来观展

【SEMI-e ·国际半导体深圳展】| 06月26-28日唯创知音语音芯片供应商 邀您来观展

世界聚焦半导体，产业规模空前！一场高端产业研学盛会即将如约而至。 SEMI-e 第六届2024国际半导体展深圳站，2024年06月26-28日将在深圳国际会展中心（宝安）开展，展会展出面积60000平方米，汇聚全国…
阅读更多...
Python | Leetcode Python题解之第155题最小栈

Python | Leetcode Python题解之第155题最小栈

题目： 题解： class MinStack:def __init__(self):self.stack []self.min_stack [math.inf]def push(self, x: int) -> None:self.stack.append(x)self.min_stack.append(min(x, self.min_stack[-1]))def pop(self) -> None:self.stack.pop()sel…
阅读更多...
Java+ffmpeg 合并两个mp4文件

Java+ffmpeg 合并两个mp4文件

使用ffmpeg测试命令 ffmpeg -i "E:\Monitor\video_20240617_10.mp4" -i "E:\Monitor\video1_20240617_10.mp4" -filter_complex "[0:v][0:a][1:v][1:a]concatn2:v1:a1[v][a]" -map "[v]" -map "[a]" -c:v libx264 -c:a…
阅读更多...
5点小技巧教你如何做一个成功的亚马逊VC卖家

5点小技巧教你如何做一个成功的亚马逊VC卖家

亚马逊Vendor Central（VC）是卖家与亚马逊直接合作的重要平台，为卖家提供了更多的销售机会和更稳定的销售渠道。那么，如何在这个平台上脱颖而出，成为一个成功的亚马逊VC卖家呢？ 以下是五个小技巧供您参考&am…
阅读更多...
NebulaGraph 知识图谱数据库使用：nebula数据库连接使用

NebulaGraph 知识图谱数据库使用：nebula数据库连接使用

参考:https://github.com/vesoft-inc/nebula-python/blob/master/example/GraphClientSimpleExample.py 文档：https://docs.nebula-graph.com.cn/3.4.1/3.ngql-guide/7.general-query-statements/2.match/#_4 pip install nebula3-python1）查询的节点&…
阅读更多...
人工智能的头号威胁：投毒攻击

人工智能的头号威胁：投毒攻击

随着掌管数字生活入口的万亿美元俱乐部企业——苹果公司跳入人工智能（AI）赛道，AI技术民主化的大幕正式拉开，同时也将AI安全问题推向舆论的风口浪尖。 根据瑞银本周一的智能手机调查报告，在中国以外的智能手机用户中&am…
阅读更多...
Vue63-配置代理-方式二

Vue63-配置代理-方式二

一、请求前缀：能灵活的控制走不走代理 1-1、请求前缀 有请求前缀的走代理服务器； 没有请求前缀的不走代理服务器。 修改代码中的请求地址，加上请求前缀 报错的原因： 解决方式： 1-2、ws配置项、changeOrigin配置项 二…
阅读更多...
redis-实战篇（1）短信登录

redis-实战篇（1）短信登录

黑马redis-实战篇笔记 开篇导读 亲爱的小伙伴们大家好，马上咱们就开始实战篇的内容了，相信通过本章的学习，小伙伴们就能理解各种redis的使用啦，接下来咱们来一起看看实战篇我们要学习一些什么样的内容 短信登录 这一块我们会使…
阅读更多...
利用LinkedHashMap实现一个LRU缓存

利用LinkedHashMap实现一个LRU缓存

一、什么是 LRU LRU是 Least Recently Used 的缩写，即最近最少使用，是一种常用的页面置换算法，选择最近最久未使用的页面予以淘汰。 简单的说就是，对于一组数据，例如：int[] a {1,2,3,4,5,6}，…
阅读更多...
Linux - 输入输出

Linux - 输入输出

一、输出格式 echo //末尾自带换行 -n //取消自带换行 -e //支持转移符 常见转义符 \n换行 \t制表符 printf // 格式化输出字符串 %-10s // %s代表字符串 -10 左对齐容纳10个字符 二、输入输出重定向 file descriptors &#x…
阅读更多...
leetcode (top100)接雨水

leetcode (top100)接雨水

题目： 题解： 可以把每个宽度看作一个桶，每个桶能接的水等于这个桶左右两个方向最高桶的最小高度再减去这个桶本身的高度。把每个桶能接的水相加即可。 难点在于如何快速找到当前桶的左右两个方向的最高桶的高度，可以先遍历一遍…
阅读更多...
LangChain入门到精通，看这这篇吊打面试官

LangChain入门到精通，看这这篇吊打面试官

导语 在人工智能领域的不断发展中，语言模型扮演着重要的角色。特别是大型语言模型（LLM），如ChatGPT，已经成为科技领域的热门话题，并受到广泛认可。在这个背景下，LangChain作为一个以LLM模型为核…
阅读更多...
什么是元数据管理？企业进行元数据管理可以满足什么目的？

什么是元数据管理？企业进行元数据管理可以满足什么目的？

元数据管理作为数据治理的重要组成部分，其作用日益凸显。元数据，即“关于数据的数据”，提供了对数据的描述、上下文和意义的详细信息，对于确保数据的准确性、一致性和可访问性至关重要。 有效的元数据管理能够帮助企业更好地理解…
阅读更多...
数位统计DP——AcWing 338. 计数问题

数位统计DP——AcWing 338. 计数问题

数位统计DP 定义 数位DP（Digital DP）是一种用于解决与数字的数位相关问题的动态规划算法。它将数字的每一位看作一个状态，通过转移状态来计算满足特定条件的数字个数或其他相关统计信息。 运用情况 统计满足特定条件的数字个数&#xff0…
阅读更多...
好用的抖音短视频矩阵系统推荐：筷子剪辑，超级编导。抖去推

好用的抖音短视频矩阵系统推荐：筷子剪辑，超级编导。抖去推

目前短视频矩阵行业如火如荼，为大家推荐几款比较好用的短视频矩阵系统。 第一款叫做筷子剪辑，由筷子科技开发，网页版应用工具，无需下载安装 主打视频剪辑，支持一键成片，视频发布等，&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023