美文网首页程序员
RabbitMQ -- part1[Hello World]

RabbitMQ -- part1[Hello World]

作者: bloke_anon | 来源:发表于2018-02-26 15:09 被阅读46次
    RabbitMQ 是一个消息中间件,它接受并转发消息。也可以认为是一个邮局,将需要投递的信件放到信箱中,确保你的收件人最终会收到此邮件。在这个过程中,RabbitMQ 充当了信箱、邮局和邮递员的角色

    1. 相关概念

    • Producing 一个发送消息的程序:
    producer
    • queue RabbitMQ中的一个信箱的名字。虽然消息流在RabbitMQ和应用程序之间传输,但只能存储在一个队列(queue)中。queue 仅受到主机内存与磁盘的限制,它本质就是个大的消息缓冲区。多个producers可以发送消息到一个队列中,并且多个consumers可以从一个队列中获取数据:
    queue
    • Consuming 主要是等待接受消息的程序:
    consumer

    2. Python Client [pika]

    • 安装: pip install pika

    3. Hello World

    hello

    上图中,"P"代表生产者(producer),"C"代表消费者(consumer),中间的box是队列(queue)--消息缓冲区。producer发送"hello"到队列中,consumer从队列中接收消息。

    • Sending
    sending

    首先编辑send.py程序去发送消息到队列中:

    #!/usr/bin/env python3
    import pika
    
    # 链接本地的RabbitMQ服务器
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    
    # 创建hello队列。在发送消息之前,首先确保队列存在,如果将消息发送至一个不存在的队列,RabbitMQ将会丢弃消息
    channel.queue_declare(queue='hello')
    
    # 发送"Hello World!"消息到"hello"队列中。在RabbitMQ中,消息不能直接发送到队列中,需要通过"exchange"将消息交到队列中。
    channel.basic_publish(exchange='',  # 空字符表示使用默认exchange
                          routing_key='hello',  # 指定队列名称
                          body='Hello World!')  # 要发送的消息
    print(" [x] Sent 'Hello World!'")
    
    # 在退出程序之前,需要确保网络缓冲区已刷新并且消息被成功提交到RabbitMQ,所以要关闭连接
    connection.close()
    

    注: 如果第一次执行之后,没有看到"Sent"消息,可能是中间件磁盘空间不足(默认最少需要200MB)导致拒绝接受消息,需要检查中间件日志来排查错误。

    • Receiving

    然后编辑receive.py程序,从队列中接受消息并且打印到终端

    #!/usr/bin/env python3
    import pika
    
    # 连接本地RabbitMQ服务器
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    
    # 如之前,需要确保队列存在,"queue_declare"操作是幂等的,可以执行多次,最终结果只是确保队列被创建
    channel.queue_declare(queue='hello')
    
    # 定义回调函数,当接收到消息后,此函数被Pika库调用
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    
    # 通知RabbitMQ从hello队列中接受到消息后调用callback回调函数
    channel.basic_consume(callback,     # 回调函数名称
                          queue='hello',    # 从哪个队列中获取消息
                          no_ack=True)      # 不需要ACK
    
    # 不停地循环等待接受数据并调用回调函数
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()
    

    在发送和接受消息中重复申明了队列 (channel.queue_declare(queue='hello')),是为了确保队列已存在。如果可以确定队列已经存在,则不必申明,例如在send.py中申明后,确保receiving.py在send.py之后启动,就不需要再次申明队列。

    • 运行

    启动consumer:

    > python3 receive.py
    [*] Waiting for messages. To exit press CTRL+C
    [x] Received 'Hello World!'
    

    启动producer,producer每次执行后会自动退出

    > python3 send.py
    [x] Sent 'Hello World!'
    
    • 查看RabbitMQ队列中有多少消息

    rabbitmqctl list_queues


    参考文档: http://www.rabbitmq.com/tutorials/tutorial-one-python.html


    相关文章

      网友评论

        本文标题:RabbitMQ -- part1[Hello World]

        本文链接:https://www.haomeiwen.com/subject/aujyxftx.html