关于python:rabbitmq基于python操作

43次阅读

共计 5645 个字符,预计需要花费 15 分钟才能阅读完成。

简略模式

 生产者
    1 链接 rabbitmq
    2 创立队列
    3 向指定的队列插入数据
消费者
    1 链接 rabbitmq
    2 监听模式
    3 确定回调函数 

示例:

  #生产者
  import pika

  #链接 rabbitmq
  creadentials = pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #创立队列
  channel.queue_declare(queue='ceshi')

  #向指定队列插入数据 [exchange: 交换机模式, 简略模式为空   routing_key: 指定队列  body: 要插入的值]
  channel.basic_publish(exchange='',routing_key='ceshi',body='hello world!')
  print('[x]--')

  #消费者
  import pika

  #创立链接
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #创立队列
  channel.queue_declare(queue='ceshi')

  #确定回调函数
  def callback(ch,method,properties,body):
  print("[x]:",body)

  #确定监听队列 [auto_ack: 默认应答]
  channel.basic_consume(queue='ceshi',auto_ack=True,on_message_callback=callback)
  print('[*]:---')
  #正式监听
  channel.start_consuming()

参数应用

1. 应答参数

 消费者确定监听队列时的一个参数
auto_ack
  True: 默认应答:从队列取走一条数据后队列里这条数据就不存在了, 如果在数据处理过程中程序呈现问题, 会造成数据失落景象
  False: 手动应答:从队列取走一条数据后, 这条数据还存在于队列中
  ch.basic_ack(delivery_tag=method.delivery_tag) 增加在回调函数最初 
  最初执行这条命令会通知队列, 我曾经执行实现了, 能够删除这条数据了
  手动应答必然会影响效率, 具体依据我的项目需要抉择:是谋求数据安全还是效率 

2. 长久化

  将数据保留到磁盘, 避免程序运行中途 rabbitmq 服务异样导致数据失落
  生产者
  创立队列时进行申明 durable=True 
  channel.queue_declare(queue='ceshi',durable=True) 
  插入数据时申明 properties=pika.BasicProperties(delivery_mode=2)
  channel.basic_publish(exchange='',
      routing_key='ceshi',
      body='hello world!',
      properties=pika.BasicProperties(delivery_mode=2)
      )
  消费者
  创立队列时进行申明 durable=True
  channel.queue_declare(queue='ceshi2',durable=True)

3. 散发参数

  轮询散发
  失常开启多个消费者都是轮询散发, 比方队列有 8 个数据, 每人 4 个
  偏心散发
  解决快的获取到的数据便越多,, 偏心散发须要应用手动应答形式才能够
  消费者增加 channel.basic_qos(prefetch_count=1)

交换机模式

公布订阅模式

  生产者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 fanout
  3 向交换机内插入数据
  消费者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 fanout
  3 创立队列并绑定交换机
  4 监听模式
  5 确定回调函数 

示例:

  #生产者
  import pika

  #链接 rabbitmq
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个名为 logs 类型为 fanout 的交换机
  channel.exchange_declare(exchange='logs',
                           exchange_type='fanout') #fanout: 公布订阅模式

  #向 logs 交换机插入数据 hello world!
  channel.basic_publish(exchange='logs',
                        routing_key='',
                        body='hello world!')
  print('[x]---')
  connection.close()

  #消费者
  import pika

  #创立链接
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个与生产者名称类型雷同的交换机, 防止先启动消费者 - 队列找不到交换机的状况
  channel.exchange_declare(exchange='logs',
                           exchange_type='fanout') #fanout: 公布订阅模式

  #创立队列   exclusive: 零碎会创立一个随机惟一的队列名
  result = channel.queue_declare(queue='',exclusive=True)
  queue_name = result.method.queue
  print(queue_name)

  #将指定队列绑定到交换机上
  channel.queue_bind(exchange='logs',
                     queue= queue_name)

  #确定回调函数
  def callback(ch,method,properties,body):
      print("[x]:",body)

  #确定监听队列 [auto_ack: 默认应答]
  channel.basic_consume(queue=queue_name,
                        auto_ack=True,
                        on_message_callback=callback)
  print('[*]:---')
  #正式监听
  channel.start_consuming()

关键字模式

  生产者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 direct
  3 向交换机内插入数据, 插入时增加关键字,routing_key: 想要进入哪一个消费者的队列, 就设置某个队列的关键字
  消费者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 direct
  3 创立队列并绑定交换机, 绑定交换机时增加关键字 routing_key
  4 监听模式
  5 确定回调函数 

示例:

  #生产者
  import pika

  #链接 rabbitmq
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个名为 logs 类型为 direct 的交换机
  channel.exchange_declare(exchange='logs',
                           exchange_type='direct') #direct: 关键字模式

  #向 logs 交换机插入数据 hello world! routing_key 为关键字
  channel.basic_publish(exchange='logs',
                        routing_key='info',
                        body='hello world!')

  print('[x]---')
  connection.close

  #消费者
  import pika

  #创立链接
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个与生产者名称类型雷同的交换机, 防止先启动消费者 - 队列找不到交换机的状况
  channel.exchange_declare(exchange='logs',
                           exchange_type='direct') #direct: 关键字模式

  #创立队列   exclusive: 零碎会创立一个随机惟一的队列名
  result = channel.queue_declare(queue='',exclusive=True)
  queue_name = result.method.queue
  print(queue_name)

  #将指定队列绑定到交换机上,routing_key: 关键字, 多个关键字绑定多个
  channel.queue_bind(exchange='logs',
                     queue= queue_name,
                     routing_key='info')
  channel.queue_bind(exchange='logs',
                     queue= queue_name,
                     routing_key='error')

  #确定回调函数
  def callback(ch,method,properties,body):
      print("[x]:",body)

  #确定监听队列 [auto_ack: 默认应答]
  channel.basic_consume(queue=queue_name,
                        auto_ack=True,
                        on_message_callback=callback)
  print('[*]:---')
  #正式监听
  channel.start_consuming()

通配符模式

  生产者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 topic
  3 向交换机内插入数据, 插入时增加关键字,routing_key: 想要进入哪一个消费者的队列, 就设置某个队列的关键字, 关键字能够应用. 来宰割
  消费者
  1 链接 rabbitmq
  2 创立一个交换机, 类型为 topic
  3 创立队列并绑定交换机, 绑定交换机时增加关键字 routing_key 关键字能够应用通配符 [*: 匹配一次,#: 匹配一次或屡次]
  4 监听模式
  5 确定回调函数 

示例:

  #生产者
  import pika

  #链接 rabbitmq
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个名为 logs 类型为 topic 的交换机
  channel.exchange_declare(exchange='logs',
                           exchange_type='topic') #topic: 通配符模式

  #向 logs 交换机插入数据 hello world!
  channel.basic_publish(exchange='logs3',
                        routing_key='usa.aaaa',
                        body='hello 21321!')

  print('[x]---')
  connection.close()

  #消费者
  import pika

  #创立链接
  pika.PlainCredentials(username='用户名',password='明码')
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='地址',port= 端口))
  channel = connection.channel()

  #申明一个与生产者名称类型雷同的交换机, 防止先启动消费者 - 队列找不到交换机的状况
  channel.exchange_declare(exchange='logs',
                           exchange_type='topic') #topic: 通配符模式

  #创立队列   exclusive: 零碎会创立一个随机惟一的队列名
  result = channel.queue_declare(queue='',exclusive=True)
  queue_name = result.method.queue
  print(queue_name)

  #将指定队列绑定到交换机上,routing_key: 关键字, 多个关键字绑定多个
  channel.queue_bind(exchange='logs',
                     queue= queue_name,
                     routing_key='#.aaaa')

  #确定回调函数
  def callback(ch,method,properties,body):
      print("[x]:",body)

  #确定监听队列 [auto_ack: 默认应答]
  channel.basic_consume(queue=queue_name,
                        auto_ack=True,
                        on_message_callback=callback)
  print('[*]:---')
  #正式监听
  channel.start_consuming()


正文完
 0