Python RabbitMQ

谁借莪１个温暖的怀抱￠ 2022-05-30 00:23 124阅读 0赞

> python pika 模块主要是用于实现消息队列的，想要使用pika，就必须先要了解一下 rabbitmq 其中mq（代表Message Queue 也就是消息队列）。是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取或者订阅队列中的消息。MQ和JMS类似，但不同的是JMS是SUN JAVA消息中间件服务的一个标准和API定义，而MQ则是遵循了AMQP协议的具体实现和产品。  
> Linux 下安装 rabbitmq 直接使用 yum install rabbitmq-server

# 启动rabbitmq #

`rabbitmq-server -detached`

> rabbitmq 启动之后默认监听的端口是 5672  
> 使用登录的名户名和密码默认都是guest，另一台机器访问rabbitmq(就是本机之外),需要打开15672端口 \#vi /etc/sysconfig/iptables 把15672端口加进去。然后重启service iptables restart

因此 安装完毕rabbitmq 之后，启动rabbitmq 服务。我们就可以使用pika 连接到 rabbitmq 实现消息推送，和获取。

pika连接 rabbitmq  
credentials = pika.PlainCredentials(‘guest’, ‘guest’)

# 这里可以连接远程IP，请记得打开远程端口 #

parameters = pika.ConnectionParameters(‘localhost’,5672,’/’,credentials)  
connection = pika.BlockingConnection(parameters)  
channel = connection.channel()

> 连接上rabbitmq 消息服务器之后，就可以使用生产者–消费者模式 一端发送消息，另外一端接收消息。 这样实现了程序直接的通信

消息生产者模型：

#! /usr/bin/env python
    
    import pika
    
    credentials = pika.PlainCredentials('guest', 'guest')
    param = pika.ConnectionParameters('localhost', 5672 '/', credentials)
    conn = pika.BlockingConnection(param)
    channel = conn.channel()
    
    number = random.randint(1,1000)
    body = 'hello world:%s' %number

> basic\_publish 是往消息队列里面推送消息，exchange表示交换器，  
> exchange能精确指定消息应该发送到哪个队列，routing\_key设置为队列的名称，body就是发送的内容

channel.basic_publish(exchange='',    
                              routing_key='hello',    
                              body=body)    
        print " [x] Sent %s" %body    
        connection.close()

> 在发送任务的时候，用delivery\_mode=2来标记任务为持久化存储  
> 如果生产者模型程序执行成功的话，消息队列里面就会有消息用rabbitmqctl命令来查看下  
> rabbitmqctl list\_queues 就会有一个hello队列，并且队列里有一条信息。接下来用receive.py来获取队列里的信息。
> 
> 消息消费者模型：  
> 和生产者模型一样，首先第一步先连接到消息服务器。然后声明一下从哪个队列获取消息  
> 接收消息更为复杂一些，需要定义一个回调函数来处理

#!/usr/bin/env python    
       import pika   
    
       credentials = pika.PlainCredentials('guest', 'nova')    
       parameters = pika.ConnectionParameters('localhost',5672,'/',credentials    )    
       connection = pika.BlockingConnection(parameters)    
       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()