python_way day12
1、RabbitMQ
2、pymysql
RabbitMQ
1、基本用法
"""
producer
""" import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.109.134'))
#创建链接 channle = connection.channel()
#创建频道 channle.queue_declare(queue='chan1')
#定义对列,当对列存在就没有意义了.因为producer和consumer谁先连接不一定 channle.basic_publish(exchange='',routing_key='chan1',body="hello world")
#1、exchange:定义exchange类型不填写默认是decodle, routing_key 就是往哪个频道发 #向chan1中发送消息 connection.close()
"""
consumer
""" import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.109.134')) channle = connection.channel()
#创建频道 channle.queue_declare(queue='chan1')
#定义对列 def callback(ch, method, properties, body):
print("recv body%r" % body) #内容
print("recv ch%r" % ch) #频道
print("recv method%r" % method)
print("recv properties" % properties) #属性 channle.basic_consume(callback, queue='chan1', no_ack=True)
#取数据,no_ack 不传确认值,这样可以增加效率,但是可能不是很稳定 channle.start_consuming()
2、保证数据不会丢失
试想一下,如果我们从队列中取完数据,但是我们还没有来得及获取数据,这个时候宕机了,但是数据被取走了,那我们改怎么办?
可以防止这些情况的数据丢失(its channel is closed, connection is closed, or TCP connection is lost)
这是就要在 【收消息一方】 一端用到no_ack了
"""
consumer
""" import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.109.134')) channle = connection.channel()
#创建频道 channle.queue_declare(queue='chan1')
#定义对列 def callback(ch, method, properties, body):
print("recv body %r %r" % (str(body,"utf8"), type(body)))
ch.basic_ack(delivery_tag = method.delivery_tag)
#ch就是处理回复确认信息,上面代码处理完我就给队列发送一条消息,如果没有给队列发送消息队列就把取走的信息再放一边 channle.basic_consume(callback, queue='chan1', no_ack=False)
#取数据,no_ack设置成False就是要回复的意思 channle.start_consuming()
#性能上要比上面不回复的差一些,但是安全性提高了
3、这是rabbitmq挂掉了怎么办?
我们可以把数据做持久化
durable
这里就需要在【发数据】一端写上一个字段,告诉rabbitmq
"""
producer
"""
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.109.134'))
#创建链接 channle = connection.channel()
#创建频道 channle.queue_declare(queue='chan1', durable=True,) #这里写上就报错,是否需要写?? #这里是队列持久化,但是用了就报错
#这里我们使用了durable =True ,意思就是在发数据的时候告诉rabbitmq chan1里的数据要持久化的保存 channle.basic_publish(exchange='',routing_key='chan1',body="hello world",properties=pika.BasicProperties(delivery_mode=2)) #alex 是消息持久化
#properties=pika.BasicProperties(delivery_mode=2) 发数据时就告诉rabbitmq 这条数据给我持久化 #向chan1中发送消息 connection.close()
4、获取消息的顺序
取消息,默认情况下取消息是按照间隔取消息,
但是这样取如果第一个实行的特别慢,二第二个执行的特别快,第二个已经执行到11了,二第一个还在执行1
第1个接收程序
第二个接收程序
如果我们获取队列中按照顺序就要指定 channel.basic_qos(prefetch_count=1) 表示谁来谁取,不再按照奇偶数排列
"""
consumer
""" import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.109.134')) channle = connection.channel()
#创建频道 channle.queue_declare(queue='chan1')
#定义对列 def callback(ch, method, properties, body):
print("recv body %r %r" % (str(body,"utf8"), type(body)))
ch.basic_ack(delivery_tag = method.delivery_tag)
#ch就是处理回复确认信息 channle.basic_qos(prefetch_count=1)
#谁来谁获取当前排到的队列中的元素 channle.basic_consume(callback, queue='chan1', no_ack=False)
#取数据,no_ack设置成False就是要回复的意思 channle.start_consuming()
5、exchange
有了exchange我们只需要向exchange发消息,让exchange判断我们要把消息发给哪一个队列
a. exchange type = fanout ,同时接收消息
这样就是和exchange关联的队列都会接收到消息
import pika connection = pika.BlockingConnection(pika.ConnectionParameters
(host='192.168.109.134'))
channle = connection.channel() #创建频道 channle.exchange_declare(exchange='logs_fanout',type='fanout')
#创建一个队列组,指定队列类型为fanout : 可以将多个队列同时绑定到一个队列组logs_fanout中 result = channle.queue_declare(exclusive=True)
queue_name = result.method.queue
#随机创建一个队列
#生成一个随机名字 channle.queue_bind(exchange='logs_fanout',queue=queue_name)
#将随机队列和队列组绑定 def callback(ch, method, propertise, body):
print(" [x] Received %r " % body) channle.basic_consume(callback,queue=queue_name,no_ack=True)
#获取这个随机队列名中的消息 channle.start_consuming()
收消息(消费者)
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.109.134'))
channel = connection.channel() channel.exchange_declare(exchange='logs_fanout',
type='fanout') message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs_fanout',
routing_key='',
body=message)
print(" [x] Sent %r" % message)
connection.close()
发消息(生产者)
#这样就可以多开几个这个 【消费者】,他们可以同时的收到信息了
b.exchange type = direct 关键字发送
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.109.134'))
channel = connection.channel() channel.exchange_declare(exchange='direct_logs',
type='direct')
#指定 exchanges type = direct result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
#指定队列名字为随机 severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1) for severity in severities:
#循环用户输入的内容
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
#routing_key=severity 绑定exchange和队列名与关键字的关系,我要监听那个队列这个队列 def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(callback,
queue=queue_name,
no_ack=True) channel.start_consuming()
收消息(消费者)
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.109.134'))
channel = connection.channel() channel.exchange_declare(exchange='direct_logs',
type='direct') severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
#关键字等于输入的第二个字段,如果argv小于1 severity就为info
message = ' '.join(sys.argv[2:]) or 'Hello World!'
#消息为argv的第二个字段或者hello world channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
#把messages往severity这个关键字中间发送 print(" [x] Sent %r:%r" % (severity, message))
connection.close()
发消息(生产者)
.
c.exchange type = topic 模糊匹配
在topic类型下,可以让队列绑定几个模糊的关键字,之后发送者将数据发送到exchange,exchange将传入”路由值“和 ”关键字“进行匹配,匹配成功,则将数据发送到指定队列。
- # 表示可以匹配 0 个 或 多个 单词
- * 表示只能匹配 一个 单词
发送者路由值 队列中
old.boy.python old.* -- 不匹配
old.boy.python old.# -- 匹配
消费者
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.109.134'))
channel = connection.channel() channel.exchange_declare(exchange='topic_logs',
type='topic') routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
发送端 (生产者)
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.109.134'))
channel = connection.channel() channel.exchange_declare(exchange='topic_logs',
type='topic') result = channel.queue_declare(exclusive=True)
# queue_name = result.method.queue
queue_name = "chan1"
#自定的名字,就这证明 topic_logs 这个通道只给chan1这个队列 topic_keys = sys.argv[1:]
if not topic_keys:
sys.stderr.write("Usage: %s [topic_key]...\n" % sys.argv[0])
sys.exit(1) for topic_key in topic_keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=topic_key) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(callback,
queue=queue_name,
no_ack=True) channel.start_consuming()
接收端(消费者)
#实验结果,如果一条信息中的 routing_key 同时被两个队列匹配,消费者是轮询拿结果的。不是同时都能拿到。这个和上面的 direct 不同
作业
二、pymysql
1、使用pymysql执行简单的sql语句
import pymysql conn = pymysql.connect(host='192.168.109.129',port=3306,user='root',passwd='hanxu',db='d1') # 创建游标
cursor = conn.cursor() # 执行SQL,并返回受影响的行数
effect_row = cursor.execute("create table d1.t3(nid int,name char(10))") # 执行SQL,并返回受影响行数
#使用占位符,第二个参数是个元组,就是放替换的元素的,
effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,)) # 执行SQL,并返回受影响行数
#同时插入多个数据
effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
print(effect_row)
#把上面这些写到内存的数据提交给数据库
conn.commit()
cursor.close()
conn.close()
2、获取新创建的自增id
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='', db='t1')
cursor = conn.cursor()
cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
effect_row = cursor.execute("create table d1.t4(nid int not null auto_increment primary key,name char(10))") #新增一个t4表,nid为主键切自增。
conn.commit()
cursor.close()
conn.close() # 获取最新自增ID
new_id = cursor.lastrowid
#因为上面新增加了2条数据,所以取到的只是最后一条的自增id
3、获取查询数据
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='', db='t1')
cursor = conn.cursor()
cursor.execute("select * from hosts") # 获取第一行数据
row_1 = cursor.fetchone() # 获取前n行数据
# row_2 = cursor.fetchmany(3) # 获取所有数据
# row_3 = cursor.fetchall()
print(row_1)
print(row_2)
print(row_3)
(1, 'han')
((2, 'xu'), (3, 'tian'), (4, 'huoa'))
None
#注意在一个代码块中每查询一次游标会向下走一行,所以当我们执行完row_1时,在执行row_2这是是从第二行开始向下打印三行的。
conn.commit()
cursor.close()
conn.close()
注:在fetch数据时按照顺序进行,可以使用cursor.scroll(num,mode)来移动游标位置,如:
- cursor.scroll(1,mode='relative') # 相对当前位置移动 1向下一个位置,-1向上一个
- cursor.scroll(2,mode='absolute') # 相对绝对位置移动
cursor=pymysql.cursors.DictCursor,以字典方式获取查询到的值
上面我们获取的表内容是以元组形式展现的,但是标签我们却不知道
import pymysql conn = pymysql.connect(host='192.168.109.129',port=3306,user='root',passwd='hanxu',db='d1') # 创建游标
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) #在创建游标的时候指定使用dict #查询
cursor.execute("select * from d1.t4") raw_all = cursor.fetchall()
print(raw_all)
[{'name': 'han', 'nid': 1}, {'name': 'xu', 'nid': 2}, {'name': 'tian', 'nid': 3}, {'name': 'huoa', 'nid': 4}, {'name': 'hua', 'nid': 5}, {'name': 'hou', 'nid': 6}]
#这样得到的值就会有标签和参数了。 conn.commit()
cursor.close()
conn.close()