Python RabbitMQ之消费者只能看到每秒的消息

我正在使用 Pika 0.98 测试 RabbitMQ 的生产者消费者示例。我的生产者在我的本地 PC 上运行，消费者在亚马逊的 EC2 实例上运行。

我的生产者坐在一个循环中，每秒发送一些系统属性。问题是我只看到消费者阅读了每 2 条消息，就好像每 2 条消息都没有被阅读一样。例如，我的生产者打印出这个(时间戳，使用的 cpu pct，使用的 RAM):

 
 
    2014-08-16 14:36:17.576000 -0700,16.0,8050806784 
    2014-08-16 14:36:18.578000 -0700,15.5,8064458752 
    2014-08-16 14:36:19.579000 -0700,15.0,8075313152 
    2014-08-16 14:36:20.580000 -0700,12.1,8074121216 
    2014-08-16 14:36:21.581000 -0700,16.0,8077778944 
    2014-08-16 14:36:22.582000 -0700,14.2,8075038720 
 

but my consumer is printing out this:

 
 
    Received '2014-08-16 14:36:17.576000 -0700,16.0,8050806784' 
    Received '2014-08-16 14:36:19.579000 -0700,15.0,8075313152' 
    Received '2014-08-16 14:36:21.581000 -0700,16.0,8077778944' 
 

The code for the producer is:

 
 
    import pika 
    import psutil 
    import time 
    import datetime 
    from dateutil.tz import tzlocal 
    import logging 
    logging.getLogger('pika').setLevel(logging.DEBUG) 
 
    connection = pika.BlockingConnection(pika.ConnectionParameters( 
            host='54.191.161.213')) 
    channel = connection.channel() 
 
    channel.queue_declare(queue='ems.data') 
 
    while True: 
        now = datetime.datetime.now(tzlocal()) 
        timestamp = now.strftime('%Y-%m-%d %H:%M:%S.%f %z') 
        msg="%s,%.1f,%d" % (timestamp, psutil.cpu_percent(),psutil.virtual_memory().used) 
        channel.basic_publish(exchange='',  
                          routing_key='ems.data', 
                          body=msg) 
        print msg 
        time.sleep(1) 
    connection.close() 
 

消费者的代码是:

 
 
    connection = pika.BlockingConnection(pika.ConnectionParameters( 
            host='0.0.0.0')) 
    channel = connection.channel() 
 
 
    channel.queue_declare(queue='hello') 
 
    print ' [*] Waiting for messages. To exit press CTRL+C' 
 
    def callback(ch, method, properties, body): 
        print " [x] Received %r" % (body,) 
 
    channel.basic_consume(callback, 
                          queue='hello', 
                          no_ack=True) 
 
    channel.start_consuming() 
 

请您参考如下方法:

您的代码在逻辑上没有问题，并且在我的机器上运行没有问题。您看到的行为表明您可能不小心启动了两个消费者，每个消费者以循环方式从队列中抓取一条消息。尝试杀死额外的消费者(如果你能找到它)，或者重新启动。