RabbitMQ基本示例,轮询机制,no_ack作用

Stella981
• 阅读 397

一、RabbitMQ简介:

'''
RabbitMQ就是消息队列
  之前不是学了Queue了吗,都是队列还学RabbitMQ干嘛?
  干的事情是一样的
  Python的Queue有两个,
     一个线程Queue生产者消费者模型,一个进程Queue用于父进程与子进程交互
     两个完全独立的Python程序就不能交互了,或者两台机器之间的Queue,Java跟Python之间不能交互了
     所以有了RabbitMQ
     
QQ跟Word通信:
    1、用socket直接通信
    2、通过硬盘通信
    3、QQ通过socket发给中间商,中间商通过socket发给Word
    第1个和第3个有啥区别呢?
    第一种直接通信比较复杂,
    第二种中间商可以省去网络通信维护的工作,而且可以实现三方或者更多方的通信
    这个中间商就叫RabbitMQ
Python语言连接RabbitMQ的模块有:
    pika主流常用、 celery分布式消息队列
'''

二、简单的示例:

RabbitMQ基本示例,轮询机制,no_ack作用 RabbitMQ基本示例,轮询机制,no_ack作用

import pika

# 相当于建立最基本的socket
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()
# 在管道里声明一个队列
channel.queue_declare(queue='q1')
# 通过管道发消息
channel.basic_publish(exchange='',
                      routing_key='q1', # 队列名
                      body='Hello World') # 消息内容
print("send: Hello World")
connection.close()

producer生产者

RabbitMQ基本示例,轮询机制,no_ack作用 RabbitMQ基本示例,轮询机制,no_ack作用

import pika

'''
消费者可能在其他机器上跨机器是没问题的
'''
# 建立连接.
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 建立管道
channel = connection.channel()
# 声明队列,为什么还要声明队列?
# 如果是生产者先运行,可以不写这个。
# 由于不确定生产者先运行还是消费者先运行,所以写上这个,随便哪个先运行都不会报错。
channel.queue_declare(queue='q1')

def callback(ch, method, properties, body):
    # ch:管道内存对象地址,method:获取前面发的routing_key或处理成功应答delivery_tag,properties:可以获取持久化模式delivery_mode
    print("-->", ch, method, properties)
    print("received:{0}".format(body))
# 开始消费消息
channel.basic_consume(callback, # 如果收到消息就调用该函数处理消息
                      queue='q1',
                      no_ack=True)
print("Waiting for messages...")
# 开始收消息了。启动就一直运行,一直收下去,没有就阻塞住
channel.start_consuming()

consumer消费者

运行结果:先启动消费者,再启动生产者(多个生产者,一个消费者)

'''
消费者控制台(启动时):
  Waiting for messages...
生产者控制台(第一次运行):
  send: Hello World
消费者控制台(变化1):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
生产者控制台(第二次运行):
  send: Hello World
消费者控制台(变化2):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
'''

 三、消息分发轮询:

上面的示例第二种运行结果:先启动消费者,再启动生产者(多个生产者,多个消费者)

'''
消费者控制台(启动消费者1):
  Waiting for messages...
消费者控制台(启动消费者2):
  Waiting for messages...
消费者控制台(启动消费者3):
  Waiting for messages...
生产者控制台(第一次运行):
  send: Hello World
消费者1控制台(变化):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
生产者控制台(第二次运行):
  send: Hello World
消费者2控制台(变化):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
生产者控制台(第三次运行):
  send: Hello World
消费者3控制台(变化):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
生产者控制台(第四次运行):
  send: Hello World
消费者1控制台(变化):
  Waiting for messages...
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'
  --> <BlockingChannel impl=<Channel number=1 OPEN....
  received:b'Hello World'轮询分发...
'''

 四、特殊情况,no_ack作用:

'''
假如现在顺序启动了3个消费者(A,B,C)处理数据,
A在处理的过程中挂掉了,怎么办呢?有两种方式:
一、no_ack设置为True:
    这种情况表示生产者不关心数据是否处理完毕。如果A挂了
   A当时的数据就丢掉了没有被处理
二、no_ack不写:
    这种情况下,如果A挂了,A的数据会转给B,  如果B挂了,A的数据会
    转给C,如果C也挂了,那么在下次启动消费者时会转给第一次启动的消费者
   这种情况下,消费者处理完毕会通知RabbitMQ,RabbitMQ就从队列里删除
   该数据。只有收到通知才从队列中删除。
'''

consumer: 改以下两处就会将所有消息完全处理

def callback(ch, method, properties, body):
    print("--->:",ch)
    time.sleep(10)
    print("received: ", body)
    # 2、加上这句,通知给RabbitMQ说这个消息已经处理完了,可以从队列里删了,如果no_ack=False这里必须要写
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback,
                      queue='q1',
                      # no_ack=True   1、将这句注释表示消费者处理一条就通知RabbitMQ
                      )

查看队列里当前消息数量:

 RabbitMQ基本示例,轮询机制,no_ack作用

点赞
收藏
评论区
推荐文章
刚刚好 刚刚好
4个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Stella981 Stella981
1年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Stella981 Stella981
1年前
Python之time模块的时间戳、时间字符串格式化与转换
Python处理时间和时间戳的内置模块就有time,和datetime两个,本文先说time模块。关于时间戳的几个概念时间戳,根据1970年1月1日00:00:00开始按秒计算的偏移量。时间元组(struct_time),包含9个元素。 time.struct_time(tm_y
Easter79 Easter79
1年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Wesley13 Wesley13
1年前
MySQL查询按照指定规则排序
1.按照指定(单个)字段排序selectfromtable_nameorderiddesc;2.按照指定(多个)字段排序selectfromtable_nameorderiddesc,statusdesc;3.按照指定字段和规则排序selec
Wesley13 Wesley13
1年前
Java日期时间API系列36
  十二时辰,古代劳动人民把一昼夜划分成十二个时段,每一个时段叫一个时辰。二十四小时和十二时辰对照表:时辰时间24时制子时深夜11:00凌晨01:0023:0001:00丑时上午01:00上午03:0001:0003:00寅时上午03:00上午0
Stella981 Stella981
1年前
Angular material mat
IconIconNamematiconcode_add\_comment_addcommenticon<maticonadd\_comment</maticon_attach\_file_attachfileicon<maticonattach\_file</maticon_attach\
Wesley13 Wesley13
1年前
PHP中的NOW()函数
是否有一个PHP函数以与MySQL函数NOW()相同的格式返回日期和时间?我知道如何使用date()做到这一点,但是我问是否有一个仅用于此的函数。例如,返回:2009120100:00:001楼使用此功能:functiongetDatetimeNow(){
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
helloworld_34035044 helloworld_34035044
7个月前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为