zoukankan      html  css  js  c++  java
  • Python RabbitMQ消息分发轮询

      1.收消息:一对多,默认依次轮询的发给每个消费端。

      2.消息确认:默认RabbitMQ不会设置no_ack=Ture,意思是,当生产者给消费者发送发送消息时,消费者处理这个消息,处理完后会手动确认发送ACK确认给服务端。

      3.如果生产端没有收到确认,这个消息就会一直轮询给每个消费端,直到生产端收到ack确认,在会把队列中的这条消息删除。

    send端
    #_*_coding:utf-8_*_
    __author__ = 'Alex Li'
    import pika,time
    
    # 实例话创建socket
    connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost'))
    
    # 声明一个管道/在管道内发消息
    channel = connection.channel()
    
    
    # 为什么再次声明queue名字:如果消费者先运行了,没有声明queue就会报错
    # 如果想要防止报错发生,就要定义queue。
    #
    # 管道内,声明一个队列,queue=queue的名字
    channel.queue_declare(queue='hello')
    
    #回调函数
    # ch 管道内存对象地址
    # method 消息发给哪个queue
    # body = 消息内容
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        time.sleep(10)
    
    # 消费消息
    # callback 如果收到消息,就调用callback函数来处理消息
    # queue 管道内的队列名字
    # no_ack = True 这条消息出没处理完都不会给服务端发确认
    channel.basic_consume(
                        callback,
                        queue='hello',)
    
    print(' [*] Waiting for messages. To exit press CTRL+C')
    
    # 启动后一直运行,没有数据会等待..
    channel.start_consuming()
    recv端 开启多个进行测试
    #_*_coding:utf-8_*_
    __author__ = 'Alex Li'
    import pika,time
    
    # 实例话创建socket
    connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost'))
    
    # 声明一个管道/在管道内发消息
    channel = connection.channel()
    
    
    # 为什么再次声明queue名字:如果消费者先运行了,没有声明queue就会报错
    # 如果想要防止报错发生,就要定义queue。
    #
    # 管道内,声明一个队列,queue=queue的名字
    channel.queue_declare(queue='hello')
    
    # 回调函数
    # ch 管道内存对象地址
    # method 消息发给哪个queue
    # body = 消息内容
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        time.sleep(10)
     # 消息处理完后会向生产端发送确认指令
     ch.basic_ack(delivery_tag=method.delivery_tag)
    
    # 消费消息
    # callback 如果收到消息,就调用callback函数来处理消息
    # queue 管道内的队列名字
    # no_ack = True 这条消息出没处理完都不会给服务端发确认
    channel.basic_consume(
                        callback,
                        queue='hello',)
    
    print(' [*] Waiting for messages. To exit press CTRL+C')
    
    # 启动后一直运行,没有数据会等待..
    channel.start_consuming()
  • 相关阅读:
    geoserver源码maven编译相关问题
    openlayers2地图控件扩展:要素删除DeleteFeature
    openlayers2地图控件扩展:图例控件LegendControl
    [小游戏资源] 微信小游戏开发资源目录
    【转】利用 three.js 开发微信小游戏的尝试
    微信小游戏开发之四:使用three.js引擎
    【转】微信小游戏开发源码_教程_工具_资源最新集合
    【转】微信小游戏学习
    微信小游戏开发Canvas资源汇总
    【转】微信小游戏开发总结
  • 原文地址:https://www.cnblogs.com/xiangsikai/p/8289348.html
Copyright © 2011-2022 走看看