python_way day12 RabbitMQ ，pymysql

973114

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='123', 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='123', 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()


三、SQLAchemy