1 概念

什么是消息队列？

基于生产者消费者模型，即生产一个消费一个，无生产则无消费，有消费无生产则等待；

采用队列数据结构，先进先出，和生活中排队一致；

为什么消息要形成队列？

计算机处理单条消息的事件始终受限，微观上来看本就逐条处理，但宏观上同时到达的任务会导致并发困难以至于积压任务；

消息队列就是对任务形成一种逻辑处理顺序，不再是无需的任务处理请求，按照先来先出来的原则，逐次处理任务；

由于微观上的逐次处理和多任务导致宏观上的并发丧失，消息队列提供了最坏的时间长度和一般的逻辑处理，计算机不再随机或者说遭遇堵塞；

对于许多任务并非要求实时性，比如大V的微博发布，对于在线用户可以直接推，但对于离线用户可以存储在消息队列，等用户上线后再推送，

对于服务器而言，减少了并发处理的任务量，对于用户而言，基本感觉不到消息的延迟，而对于推荐而言，公共区域的推送直接节省了多个用户推送的数据量，展示更加直观；

代码模板

  1. # -*- coding: UTF-8 -*-
  2. """
  3. @author:41999
  4. @file:队列.py
  5. @time:2021/10/22
  6. """
  7. import queue
  9. q = queue.Queue(maxsize=100)
  10. q.put(123)
  11. q.put(456)
  12. q.put(789)
  14. print(q.get())
  15. print(q.get())
  16. print(q.get(block=False))
  17. # block用于关闭阻塞时读取操作的挂起

为什么使用消息队列？

简介

消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ。

应用场景—外卖系统

外卖订单的数据模型：用户，时间，店铺，金额

处理流程：用户完成付款，订单系统负责提供信息，分为四个对象，

  1. 对用户而言，告知交易成功；
  2. 对骑手系统而言，告知新的任务已产生，等待抢单，
  3. 对于产品存储系统而言，告知其库存减量，
  4. 对于商家系统而言，扣除前端库存，并且告知交易产生

通知的方式：链式处理，骑手系统 —> 产品存储系统 —> 商家系统 —> 用户[完整走完四个部件，数据却可能丢失]

  1.         传统业务一对一，结算系统开启一个接口，逐次通知四个系统；
  3.         消息队列的方式：开放四个接口，被四个系统监听，并且形成一个计数器，成功读取四次后自动销毁数据，等待下一个订单的生产；
  5. 5. 消息队列充当了一个中间层，左侧是订单系统，右侧分别是骑手系统，产品库存系统，商家系统和用户，右侧四者分别订阅消息，左侧订单的变动会实时推送给右侧四个主体；

评价方式[内聚耦合]

低内聚高耦合，模块相对独立；

RabbitMQ

简介：

  1. 分为简单模式和交换机模式两种
  3.     交换机模式分为：发布订阅；关键字匹配模式；模糊匹配模式
  5. 简单模式
  7.         生产者：链接服务器；创建队列；将消息插入队列；
  9.         消费者：链接服务器；创建消息队列；

生产者代码

  1. # -*- coding: UTF-8 -*-
  2. """
  3. @author:41999
  4. @file:01producer.py
  5. @time:2021/10/27
  6. """
  7. import pika
  8. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  9. channel = connection.channel()
  11. channel.queue_declare(queue='hello')
  12. # 简单模式，交换机参数为空，第二个参数是指定队列
  13. channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
  15. print("[x] Sent 'Hello world!'")

消费者

  1. # -*- coding: UTF-8 -*-
  2. """
  3. @author:41999
  4. @file:02consumer.py
  5. @time:2021/10/27
  6. """
  7. import pika
  9. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  10. channel = connection.channel()
  12. # 创建消息队列
  13. channel.queue_declare(queue='hello')
  15. # 确定回调函数
  16. def callback(ch, mnethod, properties, body):
  17.     print("[x] Received %r" % body)
  19. # 监听队列 auto_ack 默认应答
  20. channel.basic_consume(queue='hello', auto_ack=True, on_message_callback=callback)
  22. print("[*] Waiting for messages. To exit press CTRL + C")
  23. # 正式开始监听
  24. channel.start_consuming()

最终效果

消息队列 - 图1

参数使用

应答参数

应用场景

  1. 1. 应答参数所属的函数以及消息接收阶段
  2.     1. 属于消费者模块的函数channel.basic_consume；属于最后的确定监听队列阶段，在确定回调函数之后
  3. 2. 存在的意义
  4.     1. 默认应答则在生产一个消息后消费一个消息
  5.     2. 当生产者生产一个消息，消费者创建连接--->创建队列--->创建回调函数时，程序接收但意外终止
  6.     四个步骤并没有完全走完，那么数据就会丢失[原子操作]
  7.     3. 解决策略：将应答参数改为False，手动应答

代码

  1. ===========================producer==================================================
  2. import pika
  3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  4. channel = connection.channel()
  6. channel.queue_declare(queue='hello')
  7. # 简单模式，交换机参数为空，第二个参数是指定队列
  8. channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
  10. print("[x] Sent 'Hello world!'")
  11. ===========================consumer=================================================
  12. # 确定回调函数
  13. def callback(ch, mnethod, properties, body):
  15.     print("[x] Received %r" % body)
  17. # 监听队列  auto_ack 默认应答
  18. channel.basic_consume(queue='hello', auto_ack=True, on_message_callback=callback)
  20. print("[*] Waiting for messages. To exit press CTRL + C")
  21. # 正式开始监听
  22. channel.start_consuming()

最终效果

消息队列 - 图2

持久化

应用场景

为什么设置这个参数？

RabbitMQ可能会意外停止，而持久化可以保证即使程序重启，消费者队列依然可以接收到生产者的产品

代码

  1. =====================================producer.py=====================================
  2. import pika
  3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  4. channel = connection.channel()
  6. channel.queue_declare(queue='hello', durable=True, passive=True)
  7. # 简单模式，交换机参数为空，第二个参数是指定队列
  8. channel.basic_publish(exchange='',
  9.                       routing_key='hello',
  10.                       properties=pika.BasicProperties(delivery_mode=2),
  11.                       body='Hello World!')
  13. print("[x] Sent 'Hello world!'")
  14. =====================================consumer.py=====================================
  15. import pika
  17. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  18. channel = connection.channel()
  20. # 创建消息队列
  21. channel.queue_declare(queue='hello', durable=True, passive=True)
  24. # 确定回调函数
  25. def callback(ch, mnethod, properties, body):
  26.     print("[x] Received %r" % body)
  27.     # ch.basic_ack(delivery_tag=method.delivery_tag)  # 确认消费者取得数据后，消费者在队列中删除产品
  30. # 监听队列  auto_ack 默认应答
  31. channel.basic_consume(queue='hello',
  32.                       auto_ack=True,
  33.                       on_message_callback=callback)
  35. print("[*] Waiting for messages. To exit press CTRL + C")
  36. # 正式开始监听
  37. channel.start_consuming()

报错：pika.exceptions.ChannelClosedByBroker: (406, "PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'hello' in vhost '/': received 'true' but current is 'false'")

解决

  1. 在队列建立阶段中中新增passive参数，并且设置为True。生产者与消费者均需要设置，且均为持久化
  2. 参考文档：
  3. https://www.cnblogs.com/gangzi4321/p/11001497.html

订阅者模式

场景

生产者通过交换机发布消息，多个消费者建立自己的队列从交换机读取同一份信息，类似于村里的广播

代码

  1. =======================================producer.py====================================
  2. import pika
  4. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  5. channel = connection.channel()
  7. # 声明一个名为long的，交换机类型为fanout的交换机
  8. message = "Info:20211027"
  9. channel.basic_publish(exchange='logs',
  10.                       exchange_type='fanout',
  11.                       routing_key='',
  12.                       body=message)
  13. print("[x] Sent % r" % message)
  14. connection.close()
  15. =======================================consumer.py====================================
  16. import pika
  18. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  19. channel = connection.channel()
  21. # 声明一个名为log类型为fanout的交换机
  22. channel.exchange_declare(exchange='logs',
  23.                          exchange_type='fanout')
  24. # 创建队列
  25. result = channel.queue_declare("", exclusive=True)
  26. queue_name = result.method.queue
  27. print(queue_name)
  28. # 将队列和交换机绑定
  29. channel.queue_bind(exchange='logs',
  30.                    queue=queue_name)
  31. print('[*] Waiting for logs. To exit press CTRL + C')
  34. def callback(ch, method, properties, body):
  35.     print("[X] % r" % body)
  38. # 将消费者的队列与交换机绑定
  39. channel.basic_consume(queue=queue_name,
  40.                       auto_ack=True,
  41.                       on_message_callback=callback)
  42. channel.start_consuming()

公平分发

简介

  1. 1. 缺陷
  2.     我们提供多个消费者，目的就是为了提高系统的性能，提升系统处理任务的速度;
  3.     如果将消息平均的分发给每个消费者，那么处理消息快的服务是不是会空闲下来;
  4.     而处理慢的服务可能会阻塞等待处理，这样的场景是我们不愿意看到的;
  5.     所以有了今天要说的分发模式，公平分发。
  7. 2. 能者多劳
  8.     所谓的公平分发，其实用能者多劳描述更为贴切，
  9.     根据名字就可以知道，谁有能力处理更多的任务，那么就交给谁处理，防止消息的挤压。
  11. 3. 前提
  12.     将自动应答改为手动应答

代码

  1. ===================================================producer.py===================================================
  2. import pika
  3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  4. channel = connection.channel()
  6. channel.queue_declare(queue='hello')
  7. # 简单模式，交换机参数为空，第二个参数是指定队列
  8. channel.basic_publish(exchange='',
  9.                       routing_key='hello',
  10.                       body='222')
  12. print("[x] Sent 'Hello world!'")
  13. ===================================================consumer.py===================================================
  14. import pika,time
  16. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
  17. channel = connection.channel()
  19. # 创建消息队列
  20. channel.queue_declare(queue='hello')
  23. # 确定回调函数
  24. def callback(ch, mnethod, properties, body):
  25.     time.sleep(5)
  26.     print("[x] Received %r" % body)
  27.     # ch.basic_ack(delivery_tag=method.delivery_tag)  # 确认消费者取得数据后，消费者在队列中删除产品
  30. # 公平分发---能者多劳
  31. channel.basic_qos(prefetch_count=1)
  33. # 监听队列  auto_ack 默认应答
  34. channel.basic_consume(queue='hello',
  35.                       auto_ack=False,
  36.                       on_message_callback=callback)
  38. print("[*] Waiting for messages. To exit press CTRL + C")
  39. # 正式开始监听
  40. channel.start_consuming()

问题

  1. 1. 出现的问题
  2.     1. 本次新增的参数为：channel.basic_qos(prefetch_count=1)；
  3.     2. 想要实现公平分配，即工作效率高，快速回复的对象接收最多的消息
  4. 2. 实验方法
  5.     1. 消费者分三次发送消息，间隔看手速，三个消费者持续监听同一个队列；
  6.     2. 三个消费者的睡眠时间分别为5,10,15；
  7.     3. 预期实现的效果：睡眠时间最短的消费者收到最多的消息，也就是三条；
  8.     4. 实际运行的效果：三个消费者收到的消息平摊，每人一条。

交换机模式之关键字模式

简介

发布订阅相当于群发，而关键字模式类似于私发，前者一对多，后者一对一

测试方法

对于生产者而言，只需要更改参数消息发布中的routine_key和exchange_type=’direct’ ，对于消费者，需要修改exchange_type 和 routine_key前者修改模式，后者指定关键字

一个生产者对应三个消费者

  1. ===================================================producer.py====================================================
  2. import pika
  4. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  5. channel = connection.channel()
  7. # 声明一个名为long的，交换机类型为fanout的交换机
  8. message = "Info:20211029"
  9. channel.basic_publish(exchange='logs',
  10.                       exchange_type='direct'，
  11.                       routing_key='info',
  12.                       body=message)
  14. print("[x] Sent % r" % message)
  15. connection.close()
  16. ===================================================consumer1.py===================================================
  17. import pika
  19. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  20. channel = connection.channel()
  22. # 声明一个名为log类型为fanout的交换机
  23. channel.exchange_declare(exchange='logs',
  24.                          exchange_type='direct')
  25. # 创建队列
  26. result = channel.queue_declare("", exclusive=True)
  27. queue_name = result.method.queue
  28. print(queue_name)
  29. # 将队列和交换机绑定
  30. channel.queue_bind(exchange='logs',
  31.                    queue=queue_name,
  32.                    routing_key='error',
  33.                    )
  35. print('[*] Waiting for logs. To exit press CTRL + C')
  38. def callback(ch, method, properties, body):
  39.     print("[X] % r" % body)
  42. # 将消费者的队列与交换机绑定
  43. channel.basic_consume(queue=queue_name,
  44.                       auto_ack=True,
  45.                       on_message_callback=callback)
  46. channel.start_consuming()
  47. ===================================================consumer2.py===================================================
  48. import pika
  50. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  51. channel = connection.channel()
  53. # 声明一个名为log类型为fanout的交换机
  54. channel.exchange_declare(exchange='logs',
  55.                          exchange_type='direct')
  56. # 创建队列
  57. result = channel.queue_declare("", exclusive=True)
  58. queue_name = result.method.queue
  59. print(queue_name)
  60. # 将队列和交换机绑定
  61. channel.queue_bind(exchange='logs',
  62.                    queue=queue_name,
  63.                    routing_key='warning',
  64.                    )
  65. print('[*] Waiting for logs. To exit press CTRL + C')
  68. def callback(ch, method, properties, body):
  69.     print("[X] % r" % body)
  72. # 将消费者的队列与交换机绑定
  73. channel.basic_consume(queue=queue_name,
  74.                       auto_ack=True,
  75.                       on_message_callback=callback)
  76. channel.start_consuming()
  77. ===================================================consumer3.py===================================================
  78. import pika
  80. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  81. channel = connection.channel()
  83. # 声明一个名为log类型为fanout的交换机
  84. channel.exchange_declare(exchange='logs',
  85.                          exchange_type='direct')
  86. # 创建队列
  87. result = channel.queue_declare("", exclusive=True)
  88. queue_name = result.method.queue
  89. print(queue_name)
  90. # 将队列和交换机绑定
  91. channel.queue_bind(exchange='logs',
  92.                    queue=queue_name,
  93.                    routing_key='info',
  94.                    )
  95. print('[*] Waiting for logs. To exit press CTRL + C')
  98. def callback(ch, method, properties, body):
  99.     print("[X] % r" % body)
  102. # 将消费者的队列与交换机绑定
  103. channel.basic_consume(queue=queue_name,
  104.                       auto_ack=True,
  105.                       on_message_callback=callback)
  106. channel.start_consuming()

交换机之通配符

简介

  1. 订阅者模式是一对多，关键字模式是一对一，通配符类似于群组分发，只要满足某个共同规则就可以

场景

  1. 一个生产者，四个消费者
  3. 需要修改的参数：对于生产者，在声明消息队列的时候，需要指定交换机类型为topic
  5.                 对于消费者，在声明消息队列的时候，修改交换机类型为topic，修改关键字为通配符；
  1. ================================================producer.py=======================================================
  2. import pika
  4. connection = pika.BlockingConnection(pika.ConnectionParameters(
  5.         host='localhost'))
  6. channel = connection.channel()
  8. channel.exchange_declare(exchange='logs',
  9.                          exchange_type='topic')
  11. message = "info:gunpowder"
  12. channel.basic_publish(exchange='logs',
  13.                       routing_key='USA.news',
  14.                       body=message)
  15. print(" [x] Sent %r" % message)
  16. connection.close()
  17. ================================================consumer1.py======================================================
  18. import pika
  20. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  21. channel = connection.channel()
  23. # 声明一个名为log类型为fanout的交换机
  24. channel.exchange_declare(exchange='logs',
  25.                          exchange_type='topic')
  26. # 创建队列
  27. result = channel.queue_declare("", exclusive=True)
  28. queue_name = result.method.queue
  29. print(queue_name)
  30. # 将队列和交换机绑定
  31. channel.queue_bind(exchange='logs',
  32.                    queue=queue_name,
  33.                    routing_key='#.news',
  34.                    )
  36. print('[*] Waiting for logs. To exit press CTRL + C')
  39. def callback(ch, method, properties, body):
  40.     print("[X] % r" % body)
  43. # 将消费者的队列与交换机绑定
  44. channel.basic_consume(queue=queue_name,
  45.                       auto_ack=True,
  46.                       on_message_callback=callback)
  47. channel.start_consuming()
  48. ================================================consumer2.py======================================================
  49. import pika
  51. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  52. channel = connection.channel()
  54. # 声明一个名为log类型为fanout的交换机
  55. channel.exchange_declare(exchange='logs',
  56.                          exchange_type='topic')
  57. # 创建队列
  58. result = channel.queue_declare("", exclusive=True)
  59. queue_name = result.method.queue
  60. print(queue_name)
  61. # 将队列和交换机绑定
  62. channel.queue_bind(exchange='logs',
  63.                    queue=queue_name,
  64.                    routing_key='#.weather',
  65.                    )
  66. print('[*] Waiting for logs. To exit press CTRL + C')
  69. def callback(ch, method, properties, body):
  70.     print("[X] % r" % body)
  73. # 将消费者的队列与交换机绑定
  74. channel.basic_consume(queue=queue_name,
  75.                       auto_ack=True,
  76.                       on_message_callback=callback)
  77. channel.start_consuming()
  78. ================================================consumer3.py======================================================
  79. import pika
  81. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  82. channel = connection.channel()
  84. # 声明一个名为log类型为fanout的交换机
  85. channel.exchange_declare(exchange='logs',
  86.                          exchange_type='topic')
  87. # 创建队列
  88. result = channel.queue_declare("", exclusive=True)
  89. queue_name = result.method.queue
  90. print(queue_name)
  91. # 将队列和交换机绑定
  92. channel.queue_bind(exchange='logs',
  93.                    queue=queue_name,
  94.                    routing_key='USA.#',
  95.                    )
  96. print('[*] Waiting for logs. To exit press CTRL + C')
  99. def callback(ch, method, properties, body):
  100.     print("[X] % r" % body)
  103. # 将消费者的队列与交换机绑定
  104. channel.basic_consume(queue=queue_name,
  105.                       auto_ack=True,
  106.                       on_message_callback=callback)
  107. channel.start_consuming()
  108. ================================================consumer4.py======================================================
  109. import pika
  111. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  112. channel = connection.channel()
  114. # 声明一个名为log类型为fanout的交换机
  115. channel.exchange_declare(exchange='logs',
  116.                          exchange_type='topic')
  117. # 创建队列
  118. result = channel.queue_declare("", exclusive=True)
  119. queue_name = result.method.queue
  120. print(queue_name)
  121. # 将队列和交换机绑定
  122. channel.queue_bind(exchange='logs',
  123.                    queue=queue_name,
  124.                    routing_key='EU.#',
  125.                    )
  126. print('[*] Waiting for logs. To exit press CTRL + C')
  129. def callback(ch, method, properties, body):
  130.     print("[X] % r" % body)
  133. # 将消费者的队列与交换机绑定
  134. channel.basic_consume(queue=queue_name,
  135.                       auto_ack=True,
  136.                       on_message_callback=callback)
  137. channel.start_consuming()

引用与参考

RabbitMQ常用命令：

  1. # 管理命令
  2. rabbitmq-service.bat install 或 rabbitmq-service install 
  3. rabbitmq-service.bat stop 或 rabbitmq-service stop
  4. rabbitmq-service.bat start 或 rabbitmq-server start
  5. # 查看状态
  6. rabbitmqctl status
  7. # 启动网页管理界面
  8.     # 接口： localhost:15672
  9.     # 命令：rabbitmq-plugins enable rabbitmq_management

link of downloading

  1. http://www.erlang.org/downloads
  2. https://www.rabbitmq.com/download.html
  3. # 消息延迟插件
  4. https://www.rabbitmq.com/download.html
  5. # 插件应用命令
  6. rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  7. # 安装教程链接
  8. https://www.cnblogs.com/yyee/p/14281111.html

reference

RabbitMQ工作队列之公平分发消息与消息应答（ACK） - SegmentFault 思否

【更新】RabbitMQ -05 持久化 & 不公平分发 - 知乎 (zhihu.com)