python 之路12 RabbitMQ Python 操作mysql
1. RabbitMQ简介rabbitmq服务类似于mysql、apache服务，只是提供的功能不一样。rabbimq是用来提供发送消息的服务，可以用在不同的应用程序之间进行通信。
2.安装RabbitMQ Ubuntu 14.04
sudo apt-get install rabbitmq-server
安装好后，rabbitmq服务就已经启动好了。接下来看下python编写Hello World!的实例。实例的内容就是从send.py发送“Hello World!”到rabbitmq，receive.py 从rabbitmq接收send.py发送的信息
rabbitmq消息发送流程（来源rabbitmq官网）
其中P表示produce，生产者的意思，也可以称为发送者，实例中表现为send.py；C表示consumer，消费者的意思，也可以称为接收者，实例中表现为receive.py；中间红色的表示队列的意思，实例中表现为hello队列。
python使用rabbitmq服务，可以使用现成的类库pika、txAMQP或者py-amqplib，这里选择了pika。
4.安装pika
sudo apt-get install python3-pip
sudo pip3 install pika
5.发布 连接到rabbitmq服务器，因为是在本地测试，所以就用localhost就可以了。
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(
host = 'localhost'
))#链接本地rabbitmq
chan = conn.channel()#创建一个通信频道
chan.queue_declare(queue='wyx')#声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。
chan.basic_publish( #发布
exchange='',
routing_key='wyx', #路由键， 相当于字典里面的key
body='hello wyx'#发送的内容相当于字典里面的value
)
print('sent hello wyx')
6.订阅
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(host = 'localhost')) #链接本地rabbitmq
chan = conn.channel() #创建链接频道
chan.queue_declare('wyx') #如果有wyx这个消息队列，则不创建，否则创建
def callback(ch,method,properties,body):
print('revice %r' % body)#接收发布者消息函数，并打印出内容
chan.basic_consume(callback,
queue='wyx',
no_ack=True)
print('wait')
chan.start_consuming()
7.fanout类型 RabbitMQ
上面的演示，消息是依次发送给绑定到该队列的接收端。如果要广播出去，就要使用交换机，本篇演示了交换机的工作方式也就是exchange的fanout类型
8.发布者
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost')) #链接RabbitMQ
chan = conn.channel()#创建频道
chan.exchange_declare(#创建exchange，名字为logs，类型为fanout，所有队列都可以收到，如果没有log这个名字则创建
exchange='logs',
type = 'fanout'
)
mess = 'wyx' #需要发布的消息
chan.basic_publish(
exchange='logs',#exchage名字
routing_key='',#存放的键
body=mess
)
print('hah %s' % mess)
conn.close()
9.订阅者
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(
host='127.0.0.1',#链接redis
))
chan = conn.channel()#创建频道，根据链接创建
chan.exchange_declare(exchange='logs',type='fanout')#创建exchange，名字为logs，类型为fanout，所有队列都可以收到，如果没有log这个名字则创建
result = chan.queue_declare(exclusive=True)#随机创建一个队列
queue_name= result.method.queue #队列名字
chan.queue_bind(exchange='logs',queue=queue_name)#把频道和所有队列绑定
print('wait')
def callback(ch,method,properties,body):#body为接受的消息
print('haaj %r' % body)
chan.basic_consume(callback,queue=queue_name,no_ack=True)#no_ack 是否做持久话，False为做持久化，True为不做持久化
chan.start_consuming()
10. 远程结果返回
在发布端执行一个命令，订阅者执行命令，并且返回结果
11.发布者
#!/usr/bin/env python3
#coding=utf8
import pika
class Center(object):
def __init__(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))#链接本地rabbitmq
self.channel = self.connection.channel()#创建频道
#定义接收返回消息的队列
result = self.channel.queue_declare(exclusive=True)
self.callback_queue = result.method.queue
self.channel.basic_consume(self.on_response,
no_ack=True,
queue=self.callback_queue)
#定义接收到返回消息的处理方法
def on_response(self, ch, method, props, body):
self.response = body
def request(self, n):
self.response = None
#发送计算请求，并声明返回队列
self.channel.basic_publish(exchange='',
routing_key='exec_queue',
properties=pika.BasicProperties(
reply_to = self.callback_queue,
),
body=str(n))
#接收返回的数据
while self.response is None:
self.connection.process_data_events()
return self.response
center = Center()#实例化#Center类，自动执行__init__函数
print( " 请输入想要执行的命令")
mess = input('Please enter the command you want to execute').strip()
response = center.request(mess)#执行结果
print(" [.] 执行结果 %r" % (response,) )
12 订阅者
#!/usr/bin/env python3
#coding=utf8
import pika
importsubprocess
#连接rabbitmq服务器
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
#定义队列
channel.queue_declare(queue='exec_queue')
print( '等待执行命令')
#执行命令，并return
def exec_cmd(n):
t = subprocess.Popen(n,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
stdout = t.stdout.read()
stderr = t.stderr.read()
if stdout:
return stdout
else:
return stderr
#定义接收到消息的处理方法
def request(ch, method, properties, body):
print( " [.] 执行命令 (%s)"% (body,))
body = str(body,encoding='utf-8')
response = exec_cmd(body)
#将计算结果发送回控制中心
ch.basic_publish(exchange='',
routing_key=properties.reply_to,
body=str(response))
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(request, queue='exec_queue')
channel.start_consuming()
13. 使用MySQLdb操作mysql数据库，并连接数据库
#!/usr/bin/python3
# -*- coding: UTF-8 -*-
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# 使用execute方法执行SQL语句
cursor.execute("SELECT VERSION()")
# 使用 fetchone() 方法获取一条数据库。
data = cursor.fetchone()
print（"Database version : %s " % data）
# 关闭数据库连接
db.close()
14.创建数据库表
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# 如果数据表已经存在使用 execute() 方法删除表。
cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
# 创建数据表SQL语句
sql = """CREATE TABLE EMPLOYEE (
FIRST_NAMECHAR(20) NOT NULL,
LAST_NAMECHAR(20),
AGE INT,
SEX CHAR(1),
INCOME FLOAT )"""
cursor.execute(sql)
# 关闭数据库连接
db.close()
15.数据库插入操作
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# SQL 插入语句
sql = """INSERT INTO EMPLOYEE(FIRST_NAME,
LAST_NAME, AGE, SEX, INCOME)
VALUES ('Mac', 'Mohan', 20, 'M', 2000)"""
try:
# 执行sql语句
cursor.execute(sql)
# 提交到数据库执行
db.commit()
except:
# Rollback in case there is any error
db.rollback()
# 关闭数据库连接
db.close()
16.数据库查询操作
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# SQL 查询语句
sql = "SELECT * FROM EMPLOYEE \
WHERE INCOME > '%d'" % (1000)
try:
# 执行SQL语句
cursor.execute(sql)
# 获取所有记录列表
results = cursor.fetchall()
for row in results:
fname = row
lname = row
age = row
sex = row
income = row
# 打印结果
print "fname=%s,lname=%s,age=%d,sex=%s,income=%d" % \
(fname, lname, age, sex, income )
except:
print("Error: unable to fecth data")
# 关闭数据库连接
db.close()
17.数据库更新操作
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# SQL 更新语句
sql = "UPDATE EMPLOYEE SET AGE = AGE + 1
WHERE SEX = '%c'" % ('M')
try:
# 执行SQL语句
cursor.execute(sql)
# 提交到数据库执行
db.commit()
except:
# 发生错误时回滚
db.rollback()
# 关闭数据库连接
db.close()
18.删除操作
import MySQLdb
# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","TESTDB" )
# 使用cursor()方法获取操作游标
cursor = db.cursor()
# SQL 删除语句
sql = "DELETE FROM EMPLOYEE WHERE AGE > '%d'" % (20)
try:
# 执行SQL语句
cursor.execute(sql)
# 提交修改
db.commit()
except:
# 发生错误时回滚
db.rollback()
# 关闭连接
db.close()
19.事物机制，事务机制可以确保数据一致性。
# SQL删除记录语句
sql = "DELETE FROM EMPLOYEE WHERE AGE > '%d'" % (20)
try:
# 执行SQL语句
cursor.execute(sql)
# 向数据库提交
db.commit()
except:
# 发生错误时回滚
db.rollback()
页:
[1]