消息队列MQ

RabbitMQ

本文主要是介绍RabbitMQ,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

RabbitMQ

安装

docker

# 下载镜像
docker pull rabbitmq
# 运行镜像
docker run -d --name my-rabbitmq -p 4369:4369 -p 5671:5671 -p 5672:5672 -p 25672:25672 rabbitmq

windows

  1. 安装Erlang

    1. http://www.erlang.org/downloads 下载otp_winxx_{版本}.exe

    2. 安装并配置环境变量

      1. %ERLANG_HOME%\bin

  2. 安装RabbitMq

    1. https://www.rabbitmq.com/install-windows.html官网下载rabbitmq-server-{版本}.exe

关键词

  1. Producer:生产者

  2. Consumer:消费者

  3. Channel: 信道是生产者,消费者和 RabbitMQ 通信的渠道,是建立在 TCP 连接上的虚拟连接。一个 TCP 连接上可以建立成百上千个信道,通过这种方式,可以减少系统开销,提高性能。

  4. Broker: 接收客户端连接,实现 AMQP 协议的消息队列和路由功能的进程。

  5. Virtual Host: 虚拟主机的概念,类似权限控制组,一个 Virtual Host 里可以有多个 Exchange 和 Queue,权限控制的最小粒度是 Virtual Host。

  6. Exchange: 交换机,接收生产者发送的消息,并根据 Routing Key 将消息路由到服务器中的队列 Queue。

    1. ExchangeType: 交换机类型决定了路由消息的行为

      1. direct(路由模式)

      2. fanout(发布/订阅模式)

      3. topic(匹配模式)

  7. Message Queue: 消息队列,用于存储还未被消费者消费的消息,由 Headerbody 组成。

    1. Header:由生产者添加的各种属性的集合,包括 Message 是否被持久化、优先级是多少、由哪个 Message Queue 接收等

    2. body:真正需要发送的数据内容。

  8. BindingKey: 绑定关键字,将一个特定的 Exchange 和一个特定的 Queue 绑定起来。

注意:生产者、消费者、消息代理可能不在同一主机上

Hello World

  1. pip install pika

  2. 生产者发送消息至消息队列,消费者从消息队列中接收消息

  1. 生产者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的消息队列

    3. 发布消息给Broker

    4. 关闭连接

       1 # producer.py
       2 import pika
       3 
       4 # 建立连接
       5 hostname = 'xxx.xxx.xxx.xxx'
       6 credentials = pika.PlainCredentials('username','password')
       7 parameters = pika.ConnectionParameters(hostname,credentials=credentials)
       8 connection = pika.BlockingConnection(parameters)
       9 channel = connection.channel()  # 声明接口
      10 
      11 # 创建队列
      12 channel.queue_declare(queue="hello")
      13 
      14 # 消息内容
      15 content = 'Hello World'
      16 
      17 # 发布消息
      18 channel.basic_publish(exchange='', routing_key='hello', body=content)
      19 print("[x] Sent Hello World")
      20 
      21 connection.close()
  1. 消费者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的消息队列

    3. 定义接收到消息之后触发的方法

    4. 从Broker中获取消息并消费(接收消息)

       1 # consumer.py
       2 import pika
       3 ​
       4 hostname = 'xxx.xxx.xxx.xxx'
       5 credentials = pika.PlainCredentials('username','password')
       6 parameters = pika.ConnectionParameters(hostname,credentials=credentials)
       7 connection = pika.BlockingConnection(parameters)
       8 channel = connection.channel()
       9 ​
      10 # 声明使用的队列
      11 channel.queue_declare(queue="hello")
      12 ​
      13 ​
      14 # 接受到消息后触发的方法
      15 def callback(ch, method, properties, body):
      16     print("[x] Received {}".format(body))
      17 ​
      18 ​
      19 # 消费消息
      20 channel.basic_consume('hello', callback, auto_ack=True)
      21 print('[*] Waiting for messages. To exit press Ctrl+C')
      22 channel.start_consuming()
    5. 运行结果

      1. 先运行consumer.py再运行producer.py

        python consumer.py
        # => [*] Waiting for messages. To exit press CTRL+C
        # => [x] Received 'Hello World!'
        python producer.py
        # => [x] Sent 'Hello World!'

工作队列

  1. 主要思想:生产者将耗时的任务分发给多个消费者(一个生产者,一个消息队列,多个消费者)

    1. 处理资源密集型任务,并且还要等它完成

    2. 将工作封装为一个消息队列,工作在多个消费者(进程)共享

  2. 两种消息分发机制

    1. 轮询分发:将消息轮流发送给每个消费者,要等到一个消费者处理完,才把消息发送给下一个消费者(效率低)

    2. 公平分发:只要有消费者处理完,就会把消息发送给目前空闲的消费者(效率高)

      # 告诉RabbitMQ一次向消费者只发送1个消息,直到消费者发送消息确认后再发送下一个,消息确认前将下一个消息发送给空闲的消费者
      channel.basic_qos(prefetch_count=1)
  3. 消息确认

    1. 确保消费者死亡,任务也不会丢失,将任务交付给下一个消费者

    2. 在回调函数中添加消息确认,告诉生产者已经接收到消息

      def callback(ch, method, properties, body):
          print("[x] Received %r" % body)
          time.sleep(body.count(b'.'))
          print("[x] Done"
      ​
          # 告诉生产者,消费者已收到消息,一般在处理结尾发送确认 
          ch.basic_ack(delivery_tag=method.delivery_tag)
  4. 消息持久化

    1. 确保即使RabbitMQ服务器崩溃,消息也不会丢失

    2. 确保队列是永久的

      channel.queue_declare(queue='task_queue', durable=True)
    3. 确保消息是永久的

      channel.basic_publish(exchange='',
                            routing_key='task_queue',
                            body=message,
                            properties=pika.BasicProperties(
                                delivery_mode=2,  # 确保消息是持久的
                            ))
  5. 生产者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的持久化消息队列

    3. 发布持久化消息给Broker

    4. 关闭连接

       1 import pika
       2 import sys
       3 ​
       4 hostname = '127.0.0.1'
       5 parameters = pika.ConnectionParameters(hostname)
       6 connection = pika.BlockingConnection(parameters)
       7 channel = connection.channel()
       8 ​
       9 # durable = True 代表消息队列持久化存储,False 非持久化存储
      10 channel.queue_declare(queue='work_queue',durable=True)
      11 ​
      12 message = ''.join(sys.argv[1:]) or 'Hello World'
      13 ​
      14 # delivery_mode = 2 声明消息在队列中持久化,delivery_mod = 1 消息非持久化
      15 channel.basic_publish(exchange='',
      16                       routing_key='work_queue',
      17                       body=message,
      18                       properties=pika.BasicProperties(delivery_mode=2,))
      19 ​
      20 print("[x] Sent %r" % message)
      21 connection.close()
  6. 消费者代码

    1. 与RabbitMQ建立连接

    2. 声明需要使用的持久化消息队列

    3. 定义接收到消息之后触发的方法

      1. 方法结尾发送确认信息

    4. 设置消息公平分发(接收一条,处理一条,处理结束之后再接收)

    5. 从Broker中获取消息并消费(接收消息)

       1 import pika
       2 import time
       3 ​
       4 hostname = '127.0.0.1'
       5 parameters = pika.ConnectionParameters(hostname)
       6 connection = pika.BlockingConnection(parameters)
       7 channel = connection.channel()
       8 ​
       9 # durable = True 代表消息队列持久化存储,False 非持久化存储
      10 channel.queue_declare(queue='work_queue', durable=True)
      11 ​
      12 ​
      13 def callback(ch, method, properties, body):
      14     print("[x] Received %r" % body)
      15     time.sleep(body.count(b'.'))
      16     print("[x] Done")
      17      # 告诉生产者,消费者已收到消息,一般在处理结尾发送确认
      18     ch.basic_ack(delivery_tag=method.delivery_tag)
      19 ​
      20 # 直到消费者发送消息确认后再发送下一个,消息确认前将下一个消息发送给空闲的消费者
      21 channel.basic_consume('work_queue', callback)   # auto_ack=False,自动发送确认消息默认是False
      22 ​
      23 # 告诉RabbitMQ一次向消费者只发送1个消息
      24 channel.basic_qos(prefetch_count=1)
      25 ​
      26 print("[*] Waiting for messages. To exist press Ctrl+C")
      27 channel.start_consuming()

交换机

  1. 通过声明交换机的交换类型,确定消息队列的工作模式

    1. 交换类型有三种:

      1. direct:路由模式

      2. topic:匹配模式

      3. fanout:发布订阅模式

      # 声明交换机名称、类型
      channel.exchange_declare(exchange='exchange_name',
                               exchange_type='fanout') # direct or topic or fanout

发布/订阅模式

  1. 主要思想:消息经过交换机之后,交换机将收到的所有消息广播所有队列(一个生产者,一个交换机,多个消息队列)

  2. 临时队列(因为是广播,无所谓什么队列,所以可以使用临时队列)

    # 使用result记录创建的临时队列
    result = channel.queue_declare(exclusive=True)  # exclusive=True,一旦消费者关闭连接,删除临时队列
  3. 绑定(交换机需要和队列绑定)

    channel.queue_bind(exchange='exchange_name',
                       queue=result.method.queue)

    (一个生产者,一个交换机,多个消息队列,一个消息队列对应一个消费者?)

  4. 思考:一个消息队列能对应多个消费者吗?

  5. 生产者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的交换机,类型为fanout

    3. 发布消息给交换机

    4. 关闭连接

       1 # producer.py
       2 import pika
       3 import sys
       4 ​
       5 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
       6 channel = connection.channel()
       7 ​
       8 # 声明交换机是广播类型的
       9 channel.exchange_declare(exchange='logs', exchange_type='fanout')
      10 ​
      11 message = ' '.join(sys.argv[1:]) or 'info: Hello World'
      12 ​
      13 channel.basic_publish(exchange='logs', routing_key='', body=message)
      14 ​
      15 print('[x] Sent %r ' % message)
      16 connection.close()
  6. 消费者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的交换机,类型为fanout

    3. 声明创建临时队列

    4. 获取临时队列名

    5. 绑定临时队列

    6. 定义接收到消息之后触发的方法

    7. 从Broker中获取消息并消费

       1 # consumer.py
       2 import pika
       3 import sys
       4 ​
       5 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
       6 channel = connection.channel()
       7 ​
       8 # 声明交换机
       9 channel.exchange_declare(exchange='logs', exchange_type='fanout')
      10 # 声明只允许当前连接的队列
      11 result = channel.queue_declare('',exclusive=True)
      12 # 获取临时队列名
      13 queue_name = result.method.queue
      14 channel.queue_bind(exchange='logs', queue=queue_name)
      15 print('[*] Waiting for logs. To exit press CTRL+C')
      16 ​
      17 ​
      18 def callback(ch, method, properties, body):
      19     print('[x] %r ' % body)
      20 ​
      21     
      22 channel.basic_consume('', callback)
      23 channel.start_consuming()

路由模式

  1. 主要思想:为了让生产者发布的消息定向的精确发送到指定的队列(严格过滤,完全匹配)

  2. 在绑定的时候使用routing_key参数,即绑定键

    channel.queue_bind(exchange=exchange_name,
                       queue=queue_name,
                       routing_key='black')
  3. 也可以使用多个绑定实现类广播效果(当然之后的匹配模式就是如此)

    例如日志消息

  4. 生产者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的交换机,类型为direct

    3. 通过命令行第2个参数确定交换机的routing_key

    4. 发布消息给交换机

    5. 关闭连接

       1 import sys
       2 import pika
       3 ​
       4 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
       5 channel = connection.channel()
       6 ​
       7 channel.exchange_declare(exchange='logs_direct', exchange_type='direct')
       8 ​
       9 print(sys.argv[1:])
      10 # 根据命令行输入的第2个参数确定routing_key(error,warning,info)
      11 # 将不同类型的日志消息保存到不同的队列
      12 severity = sys.argv[1] if len(sys.argv) > 2 else 'info'
      13 message = ' '.join(sys.argv[2:]) or 'Hello World'
      14 ​
      15 channel.basic_publish(exchange='logs_direct', routing_key=severity, body=message)
      16 ​
      17 print("[x] Sent %r:%r " % (severity, message))
      18 ​
      19 connection.close()
  5. 消费者代码

    1. 与RabbitMQ建立连接

    2. 声明使用的交换机,类型为direct

    3. 声明创建临时队列

    4. 获取临时队列名

    5. 绑定临时队列,通过命令行的参数确定交换机的routing_key

    6. 定义接收到消息之后触发的方法

    7. 从Broker中获取消息并消费

      import pika
      import sys
      ​
      connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
      channel = connection.channel()
      ​
      # 设置路由模式
      channel.exchange_declare(exchange='logs_direct', exchange_type='direct')
      ​
      # 声明临时队列
      result = channel.queue_declare('', exclusive=True)
      queue_name = result.method.queue
      ​
      # 设置队列和交换机绑定的键,通过参数确定接收什么类型的日志消息(即从指定的队列中获取消息)
      severties = sys.argv[1:]
      for severity in severties:
          channel.queue_bind(queue=queue_name, exchange='logs_direct', routing_key=severity)
      ​
      print("[*] Waiting for logs. To exit press CTRL+C")
      ​
      ​
      def callback(ch, method, properties, body):
          print("[x] %r:%r" % (method.routing_key, body))
      ​
      ​
      channel.basic_consume(queue_name, callback)
      ​
      channel.start_consuming()
    8. 运行

      # 如果只想保存'warning'和'error'(而不是'info')将消息记录到文件中,只需打开一个控制台并输入:
      python receive_logs_direct.py warning error > logs_from_rabbit.log
      ​
      # 如果您希望在屏幕上看到所有日志消息,请打开一个新终端并执行以下操作:
      python receive_logs_direct.py info warning error
      ​
      # 例如,要输出error日志消息,只需输入:
      python emit_log_direct.py error "Run. Run. Or it will explode."

匹配模式

  1. 主要思想:路由模式是精确严格的筛选,而匹配模式根据匹配的条件过滤(模糊匹配)

  2. 规范

    1. 必须是由”.“连接单词列表

    2. 最多255个字节

    3. ”*“ 可以代替一个单词

    4. ”#“可以代替0个或者多个单词

  3. 交换机类型为topic

  4. 生产者代码

     1 # emit_log_topic.py
     2 import pika
     3 
     4 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
     5 channel = connection.channel()
     6 
     7 # 交换机类型为topic
     8 channel.exchange_declare(exchange='logs_topic', exchange_type='topic')
     9 
    10 print(sys.argv[1:])
    11 severity = sys.argv[1] if len(sys.argv) > 2 else 'anonymous.info'
    12 message = ' '.join(sys.argv[2:]) or 'Hello World'
    13 
    14 channel.basic_publish(exchange='logs_topic', routing_key=severity, body=message)
    15 
    16 print("[x] Sent %r:%r " % (severity, message))
    17 
    18 connection.close()
  5. 消费者代码

     1 # receive_logs_topic.py
     2 import pika
     3 import sys
     4 
     5 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
     6 channel = connection.channel()
     7 
     8 # 交换机类型为topic
     9 channel.exchange_declare(exchange='logs_topic', exchange_type='topic')
    10 
    11 result = channel.queue_declare('', exclusive=True)
    12 queue_name = result.method.queue
    13 
    14 # 设置队列和交换机绑定的键
    15 severties = sys.argv[1:]
    16 for severity in severties:
    17     channel.queue_bind(queue=queue_name, exchange='logs_topic', routing_key=severity)
    18 
    19 print("[*] Waiting for logs. To exit press CTRL+C")
    20 
    21 
    22 def callback(ch, method, properties, body):
    23     print("[x] %r:%r" % (method.routing_key, body))
    24 
    25 
    26 channel.basic_consume(queue_name, callback)
    27 
    28 channel.start_consuming()

     

  6. 运行

    #要接收所有日志运行:
    python receive_logs_topic.py "#"
    ​
    #要从设施“ kern ” 接收所有日志:
    python receive_logs_topic.py "kern.*"
    ​
    #或者,如果您只想听到关于“ critical ”日志的信息:
    python receive_logs_topic.py "*.critical"
    ​
    #您可以创建多个绑定:
    python receive_logs_topic.py "kern." ".critical"
    ​
    #发布带有路由键“ kern.critical ”类型的日志:
    python emit_log_topic.py "kern.critical" "A critical kernel error"

远程过程调用(RPC)

  1. 主要思想:客户端与服务器之间是完全解耦的,即两端既是消息的发送者也是接受者。

  2. 服务器代码(消费者)

    1. 建立连接

    2. 声明队列

    3. 定义接收请求的回调函数(函数处理之后,向Client发送响应)

      1. properties.reply_to:客户端发来的绑定键

      2. properties.correlation_id:客户端发来的相关ID

    4. 设置公平分发

    5. 接收消息,触发请求

       1 # server.py
       2 import pika
       3 ​
       4 hostname = '192.168.253.129'
       5 credentials = pika.PlainCredentials('admin', 'admin')
       6 parameters = pika.ConnectionParameters(hostname, credentials=credentials)
       7 connecion = pika.BlockingConnection(parameters=parameters)
       8 channel = connecion.channel()
       9 ​
      10 channel.queue_declare(queue='rpc_queue')
      11 ​
      12 ​
      13 def fib(n):
      14     """
      15     返回第n个斐波那契数
      16     :param n:返回数量
      17     :return: 返回第n个斐波那契数
      18     """
      19     if n == 0:
      20         return 0
      21     elif n == 1:
      22         return 1
      23     else:
      24         return fib(n - 1) + fib(n - 2)
      25 ​
      26 ​
      27 def on_request(ch, method, properties, body):
      28     """
      29     接收client发来的请求,处理之后,再向client发送响应信息
      30     :param ch:
      31     :param method:
      32     :param properties:client请求的属性
      33     :param body:响应信息的内容
      34     :return:
      35     """
      36 ​
      37     # 收到请求后的处理过程
      38     n = int(body)
      39     print(f'[.] fib({n})')
      40     response = fib(n)
      41 ​
      42     # 将处理结果发送回client
      43     channel.basic_publish(exchange='',
      44                           routing_key=properties.reply_to,  # 响应的信息是要发回给请求的队列,而不是rpc_queue
      45                           properties=pika.BasicProperties(
      46                               correlation_id=properties.correlation_id  # 请求的唯一值属性
      47                           ),
      48                           body=str(response))
      49     ch.basic_ack(delivery_tag=method.delivery_tag)  # 消息确认
      50 ​
      51 ​
      52 channel.basic_qos(prefetch_count=1)  # 处理过程中只接受一条消息
      53 channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)    # 接受Client请求
      54 print(" [x] Awaiting RPC requests")
      55 channel.start_consuming()
    
    
  3. 客户端代码(生产者)

    1. 初始化

      1. 建立连接

      2. 回调处理响应消息

    2. 定义处理响应消息的回调函数

    3. 定义发送请求的函数

    4. 创建对象

    5. 发送请求

    6. 关闭连接

       1 # client.py
       2 import pika
       3 import uuid
       4 ​
       5 ​
       6 class FibRPCClient:
       7     '''
       8     定义一个客户端类,发送信息的Producer
       9     初始化的时候建立连接和渠道
      10     随机生成一个队列作为回调队列
      11     从回调队列中接收server响应的消息
      12     '''
      13 ​
      14     def __init__(self):
      15         hostname = '192.168.253.129'
      16         credentials = pika.PlainCredentials('admin', 'admin')
      17         parameters = pika.ConnectionParameters(hostname, credentials=credentials)
      18         self.connecion = pika.BlockingConnection(parameters=parameters)
      19         self.channel = self.connecion.channel()
      20         result = self.channel.queue_declare(queue='',exclusive=True) # 声明临时的队列
      21         self.callback_queue = result.method.queue  # 记录临时队列的队列号
      22         self.channel.basic_consume(queue=self.callback_queue,on_message_callback=self.on_response)  # 接受响应信息
      23         self.response = ''
      24         self.correlation_id = ''
      25 ​
      26     def on_response(self, ch, method, properties, body):
      27         """
      28         处理响应信息
      29         :param ch:
      30         :param method:
      31         :param properties:从属性中获取correlation_id
      32         :param body: 获取响应信息的内容
      33         :return:
      34         """
      35         # 当请求的唯一值相同则获取响应信息
      36         if self.correlation_id == properties.correlation_id:
      37             self.response = body
      38 ​
      39     def call(self, n):
      40         """
      41         Fib客户端发送请求
      42         :param n:
      43         :return:
      44         """
      45         self.response = None
      46         self.correlation_id = str(uuid.uuid4())    # 唯一的ID
      47         # 发送请求到server端
      48         self.channel.basic_publish(exchange='',
      49                                    routing_key='rpc_queue',
      50                                    properties=pika.BasicProperties(
      51                                        reply_to=self.callback_queue, # 指定回调队列
      52                                        correlation_id=self.correlation_id  # 使请求唯一
      53                                    ),
      54                                    body=str(n))
      55         # 等待server的响应
      56         while self.response is None:
      57             self.connecion.process_data_events()
      58 ​
      59         return int(self.response)
      60 ​
      61 ​
      62 fib_rpc = FibRPCClient()
      63 print('[x] Requesting fib(30)')
      64 response = fib_rpc.call(30)
      65 print(f'[.] Got %s ' % response)
      66 fib_rpc.connecion.close()
这篇关于RabbitMQ的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!