一、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个有啥区别呢？    第一种直接通信比较复杂，    第二种中间商可以省去网络通信维护的工作，而且可以实现三方或者更多方的通信    这个中间商就叫RabbitMQPython语言连接RabbitMQ的模块有：    pika主流常用、 celery分布式消息队列'''

二、简单的示例：

import pika# 相当于建立最基本的socketconnection = 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()

 

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()

 

运行结果：先启动消费者，再启动生产者（多个生产者，一个消费者）

'''消费者控制台(启动时)：  Waiting for messages...生产者控制台(第一次运行)：  send: Hello World消费者控制台(变化1)：  Waiting for messages...  --> 
         
          
         
        

 三、消息分发轮询：

上面的示例第二种运行结果：先启动消费者，再启动生产者（多个生产者，多个消费者）

'''消费者控制台(启动消费者1)：  Waiting for messages...消费者控制台(启动消费者2)：  Waiting for messages...消费者控制台(启动消费者3)：  Waiting for messages...生产者控制台(第一次运行)：  send: Hello World消费者1控制台(变化)：  Waiting for messages...  --> 
         
          
           
            
           
          
         
        

 四、特殊情况，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                      )

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