python RabbitMQ队列使用

关于python的queue介绍

• 关于python的队列，内置的有两种，一种是线程queue，另一种是进程queue，但是这两种queue都是只能在同一个进程下的线程间或者父进程与子进程之间进行队列通讯，并不能进行程序与程序之间的信息交换，这时候我们就需要一个中间件，来实现程序之间的通讯。

RabbitMQ

• MQ并不是python内置的模块，而是一个需要你额外安装（ubunto可直接apt-get其余请自行百度。）的程序，安装完毕后可通过python中内置的pika模块来调用MQ发送或接收队列请求。接下来我们就看几种python调用MQ的模式（作者自定义中文形象的模式名称）与方法。

RabbitMQ设置远程链接账号密码

启动rabbitmq web服务：

远程访问rabbitmq:自己增加一个用户，步骤如下：

• l1. 创建一个admin用户：sudo rabbitmqctl add_user admin 123123

• l2. 设置该用户为administrator角色：sudo rabbitmqctl set_user_tags admin administrator

• l3. 设置权限：sudo rabbitmqctl set_permissions -p '/' admin '.' '.' '.'

• l4. 重启rabbitmq服务：sudo service rabbitmq-server restart

之后就能用admin用户远程连接rabbitmq server了。

轮询消费模式

此模式下，发送队列的一方把消息存入mq的指定队列后，若有消费者端联入相应队列，即会获取到消息，并且队列中的消息会被消费掉。

若有多个消费端同时连接着队列，则会已轮询的方式将队列中的消息消费掉。

接下来是代码实例：

producer生产者

# !/usr/bin/env python
  import pika
  credentials = pika.PlainCredentials('admin','123456')
  connection = pika.BlockingConnection(pika.ConnectionParameters(
'192.168.56.19',5672,'/',credentials))
  channel = connection.channel()

  # 声明queue
   channel.queue_declare(queue='balance')

  # n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
    channel.basic_publish(exchange='',
                  routing_key='balance',
                  body='Hello World!')
  print(" [x] Sent 'Hello World!'")
  connection.close()

发送过队列后，可在MQ服务器中查看队列状态

  [root@localhost ~]# rabbitmqctl list_queues

  Listing queues ...

  hello    1

consumer消费者

# _*_coding:utf-8_*_
  __author__ = 'Alex Li'
  import pika

  credentials = pika.PlainCredentials('admin','123456')
  connection = pika.BlockingConnection(pika.ConnectionParameters(
'192.168.56.19',5672,'/',credentials))
  channel = connection.channel()

  # You may ask why we declare the queue again ‒ we have already declared it in our previous code.
  # We could avoid that if we were sure that the queue already exists. For example if send.py program
  # was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
   channel.queue_declare(queue='balance')


  def callback(ch, method, properties, body):
  print(" [x] Received %r" % body)


  channel.basic_consume(callback,
                  queue='balance',
                  no_ack=True)

  print(' [*] Waiting for messages. To exit press CTRL+C')
  channel.start_consuming()

接收队列后，查看一下队列状态

  [root@localhost ~]#  rabbitmqctl list_queues

  Listing queues ...

  hello    0

import pika

credentials = pika.PlainCredentials('wu','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1',5672,'simple',credentials))
channel = connection.channel() #在连接上创建一个频道

channel.queue_declare(queue='pikamq') #声明一个队列，生产者和消费者都要声明一个相同的队列，用来防止万一某一方挂了，另一方能正常运行

channel.basic_publish(exchange='', #交换机
                  routing_key='pikamq',  # queue名字 #路由键，写明将消息发往哪个队列，本例是将消息发往队列pikamq
                  body='Test Message') # 消息内容
connection.close() #当生产者发送完消息后，可选择关闭连接

生产者

import pika
import random
  
credentials = pika.PlainCredentials('guest', 'geust')
#这里可以连接远程IP，请记得打开远程端口  
parameters = pika.ConnectionParameters('localhost',5672,'/',credentials)  
connection = pika.BlockingConnection(parameters)  
channel = connection.channel()  
  
#channel.queue_declare(queue='hello')  
number = random.randint(1,1000)
body = 'hello world:%s' %number
channel.basic_publish(exchange='',  
                      routing_key='hello',  
                      body=body)  
print " [x] Sent %s" %body  
connection.close()  

消费者

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

https://www.cnblogs.com/kerwinC/p/5967584.html

生产者

import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost'))#建立一个最基本的socket
chanel = connection.channel()#声明一个管道

chanel.queue_declare(queue='name')#给管道创建一个队列，参数是管道队列名。

chanel.basic_publish(exchange='',
                 routing_key='name',
                 body ='HELLO WORD!')#要发送的消息。
print( '发出一个消息')
connection.close()#关闭

消费者

import pika,time

consumer = pika.BlockingConnection\
(pika.ConnectionParameters('localhost'))#创建socket连接
channel = consumer.channel()#创建管道
channel.queue_declare(queue='name')

def backcall(ch,method,properties,body):#参数body是发送过来的消息。
    print(ch,method,properties)
    time.sleep(15)
    print('[x] Received %r'%body)

channel.basic_consume(backcall,#回调函数。执行结束后立即执行另外一个函数返回给发送端是否执行完毕。
                  queue='name',
                  no_ack=True#不会告知服务端我是否收到消息。一般注释。
                   )#如果注释掉，对方没有收到消息的话不会将消息丢失，始终在队列里等待下次发送。

print('waiting for message To exit   press CTRL+C')
channel.start_consuming()#启动后进入死循环。一直等待消息。