一、RabbitMQ简介：
1''' 2RabbitMQ就是消息队列 3 之前不是学了Queue了吗，都是队列还学RabbitMQ干嘛？ 4 干的事情是一样的 5 Python的Queue有两个， 6 一个线程Queue生产者消费者模型，一个进程Queue用于父进程与子进程交互 7 两个完全独立的Python程序就不能交互了，或者两台机器之间的Queue，Java跟Python之间不能交互了 8 所以有了RabbitMQ 9 10QQ跟Word通信： 11 1、用socket直接通信 12 2、通过硬盘通信 13 3、QQ通过socket发给中间商，中间商通过socket发给Word 14 第1个和第3个有啥区别呢？ 15 第一种直接通信比较复杂， 16 第二种中间商可以省去网络通信维护的工作，而且可以实现三方或者更多方的通信 17 这个中间商就叫RabbitMQ 18Python语言连接RabbitMQ的模块有： 19 pika主流常用、 celery分布式消息队列 20'''
二、简单的示例：
1import pika 2 3# 相当于建立最基本的socket 4connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 5# 声明一个管道 6channel = connection.channel() 7# 在管道里声明一个队列 8channel.queue_declare(queue='q1') 9# 通过管道发消息 10channel.basic_publish(exchange='', 11 routing_key='q1', # 队列名 12 body='Hello World') # 消息内容 13print("send: Hello World") 14connection.close()
producer生产者
1import pika 2 3''' 4消费者可能在其他机器上跨机器是没问题的 5''' 6# 建立连接. 7connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8# 建立管道 9channel = connection.channel() 10# 声明队列,为什么还要声明队列？ 11# 如果是生产者先运行，可以不写这个。 12# 由于不确定生产者先运行还是消费者先运行，所以写上这个，随便哪个先运行都不会报错。 13channel.queue_declare(queue='q1') 14 15def callback(ch, method, properties, body): 16 # ch：管道内存对象地址，method：获取前面发的routing_key或处理成功应答delivery_tag，properties：可以获取持久化模式delivery_mode 17 print("-->", ch, method, properties) 18 print("received:{0}".format(body)) 19# 开始消费消息 20channel.basic_consume(callback, # 如果收到消息就调用该函数处理消息 21 queue='q1', 22 no_ack=True) 23print("Waiting for messages...") 24# 开始收消息了。启动就一直运行，一直收下去，没有就阻塞住 25channel.start_consuming()
consumer消费者
运行结果：先启动消费者，再启动生产者（多个生产者，一个消费者）
1''' 2消费者控制台(启动时)： 3 Waiting for messages... 4生产者控制台(第一次运行)： 5 send: Hello World 6消费者控制台(变化1)： 7 Waiting for messages... 8 --> <BlockingChannel impl=<Channel number=1 OPEN.... 9 received:b'Hello World' 10生产者控制台(第二次运行)： 11 send: Hello World 12消费者控制台(变化2)： 13 Waiting for messages... 14 --> <BlockingChannel impl=<Channel number=1 OPEN.... 15 received:b'Hello World' 16 --> <BlockingChannel impl=<Channel number=1 OPEN.... 17 received:b'Hello World' 18'''
三、消息分发轮询：
上面的示例第二种运行结果：先启动消费者，再启动生产者（多个生产者，多个消费者）
1''' 2消费者控制台(启动消费者1)： 3 Waiting for messages... 4消费者控制台(启动消费者2)： 5 Waiting for messages... 6消费者控制台(启动消费者3)： 7 Waiting for messages... 8生产者控制台(第一次运行)： 9 send: Hello World 10消费者1控制台(变化)： 11 Waiting for messages... 12 --> <BlockingChannel impl=<Channel number=1 OPEN.... 13 received:b'Hello World' 14生产者控制台(第二次运行)： 15 send: Hello World 16消费者2控制台(变化)： 17 Waiting for messages... 18 --> <BlockingChannel impl=<Channel number=1 OPEN.... 19 received:b'Hello World' 20生产者控制台(第三次运行)： 21 send: Hello World 22消费者3控制台(变化)： 23 Waiting for messages... 24 --> <BlockingChannel impl=<Channel number=1 OPEN.... 25 received:b'Hello World' 26生产者控制台(第四次运行)： 27 send: Hello World 28消费者1控制台(变化)： 29 Waiting for messages... 30 --> <BlockingChannel impl=<Channel number=1 OPEN.... 31 received:b'Hello World' 32 --> <BlockingChannel impl=<Channel number=1 OPEN.... 33 received:b'Hello World'轮询分发... 34'''
四、特殊情况，no_ack作用：
1''' 2假如现在顺序启动了3个消费者（A，B，C）处理数据， 3A在处理的过程中挂掉了，怎么办呢？有两种方式： 4一、no_ack设置为True： 5 这种情况表示生产者不关心数据是否处理完毕。如果A挂了 6 A当时的数据就丢掉了没有被处理 7二、no_ack不写： 8 这种情况下，如果A挂了，A的数据会转给B， 如果B挂了，A的数据会 9 转给C，如果C也挂了，那么在下次启动消费者时会转给第一次启动的消费者 10 这种情况下，消费者处理完毕会通知RabbitMQ，RabbitMQ就从队列里删除 11 该数据。只有收到通知才从队列中删除。 12'''
consumer: 改以下两处就会将所有消息完全处理
1def callback(ch, method, properties, body): 2 print("--->:",ch) 3 time.sleep(10) 4 print("received: ", body) 5 # 2、加上这句，通知给RabbitMQ说这个消息已经处理完了，可以从队列里删了，如果no_ack=False这里必须要写 6 ch.basic_ack(delivery_tag=method.delivery_tag) 7 8channel.basic_consume(callback, 9 queue='q1', 10 # no_ack=True 1、将这句注释表示消费者处理一条就通知RabbitMQ 11 )
查看队列里当前消息数量：
