Python操作Redis、Memcache、RabbitMQ、SQLAlchemy详解编程语言

class ConnectionPool(object):  
    ........... 
  def __init__(self, connection_class=Connection, max_connections=None, 
         **connection_kwargs):  #类初始化时调用构造函数 
    max_connections = max_connections or 2 ** 31 
    if not isinstance(max_connections, (int, long)) or max_connections < 0: #判断输入的max_connections是否合法 
      raise ValueError('"max_connections" must be a positive integer') 
    self.connection_class = connection_class #设置对应的参数 
    self.connection_kwargs = connection_kwargs 
    self.max_connections = max_connections 
    self.reset() #初始化ConnectionPool 时的reset操作 
  def reset(self): 
    self.pid = os.getpid() 
    self._created_connections = 0 #已经创建的连接的计数器 
    self._available_connections = []  #声明一个空的数组，用来存放可用的连接 
    self._in_use_connections = set() #声明一个空的集合，用来存放已经在用的连接 
    self._check_lock = threading.Lock() 
....... 
  def get_connection(self, command_name, *keys, **options): #在连接池中获取连接的方法 
    "Get a connection from the pool" 
    self._checkpid() 
    try: 
      connection = self._available_connections.pop() #获取并删除代表连接的元素，在第一次获取connectiong时，因为_available_connections是一个空的数组， 
      会直接调用make_connection方法 
    except IndexError: 
      connection = self.make_connection() 
    self._in_use_connections.add(connection)  #向代表正在使用的连接的集合中添加元素 
    return connection   
  def make_connection(self): #在_available_connections数组为空时获取连接调用的方法 
    "Create a new connection" 
    if self._created_connections >= self.max_connections:  #判断创建的连接是否已经达到最大限制，max_connections可以通过参数初始化 
      raise ConnectionError("Too many connections") 
    self._created_connections += 1  #把代表已经创建的连接的数值+1 
    return self.connection_class(**self.connection_kwargs)   #返回有效的连接，默认为Connection(**self.connection_kwargs) 
  def release(self, connection): #释放连接，链接并没有断开，只是存在链接池中 
    "Releases the connection back to the pool" 
    self._checkpid() 
    if connection.pid != self.pid: 
      return 
    self._in_use_connections.remove(connection)  #从集合中删除元素 
    self._available_connections.append(connection) #并添加到_available_connections 的数组中 
  def disconnect(self): #断开所有连接池中的链接 
    "Disconnects all connections in the pool" 
    all_conns = chain(self._available_connections, 
             self._in_use_connections) 
    for connection in all_conns: 
      connection.disconnect()

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
import redis 
class RedisHelper: 
 def __init__(self): 
        self.__conn = redis.Redis(host='127.0.0.1')  #连接redis服务器 
        self.chan_sub = 'fm104.5'        #订阅频道 
        self.chan_pub = 'fm104.5'        #发布频道 
 def public(self, msg):                  #定义发布函数，msg为发布消息 
        self.__conn.publish(self.chan_pub, msg) 
        return True 
 def subscribe(self): 
        pub = self.__conn.pubsub()        #定义接收函数，接收消息 
        pub.subscribe(self.chan_sub) 
        pub.parse_response()               #等待消息 
        return pub

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
import Queue 
import threading 
message = Queue.Queue(10)  #调用队列生成对象 
def producer(i):     
    while True:         
        message.put(i)        #存放一个值到队列 
def consumer(i):     
    while True:         
        msg = message.get() #获取队列的第一个值 
for i in range(12):     
    t = threading.Thread(target=producer, args=(i,))     
    t.start() 
for i in range(10):     
    t = threading.Thread(target=consumer, args=(i,))     
    t.start()

import pika 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host=192.168.21.128)) 
channel = connection.channel() 
 
channel.queue_declare(queue='hello') 
 
def callback(ch, method, properties, body): 
    print(" [x] Received %r" % body) 
    import time 
    time.sleep(10) 
    print 'ok' 
    ch.basic_ack(delivery_tag = method.delivery_tag) 
 
channel.basic_consume(callback, 
                      queue='hello', 
                      no_ack=False) 
 
print(' [*] Waiting for messages. To exit press CTRL+C') 
channel.start_consuming() 
 
消费者

#!/usr/bin/env python 
import pika 
 
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.21.128')) 
channel = connection.channel() 
 
# make message persistent 
channel.queue_declare(queue='hello', durable=True) 
 
channel.basic_publish(exchange='', 
                      routing_key='hello', 
                      body='Hello World!', 
                      properties=pika.BasicProperties( 
                          delivery_mode=2, # make message persistent 
                      )) 
print(" [x] Sent 'Hello World!'") 
connection.close()

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
import pika 
 
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.21.128')) 
channel = connection.channel() 
 
# make message persistent 
channel.queue_declare(queue='hello', durable=True) 
 
 
def callback(ch, method, properties, body): 
    print(" [x] Received %r" % body) 
    import time 
    time.sleep(10) 
    print 'ok' 
    ch.basic_ack(delivery_tag = method.delivery_tag) 
 
channel.basic_consume(callback, 
                      queue='hello', 
                      no_ack=False) 
 
print(' [*] Waiting for messages. To exit press CTRL+C') 
channel.start_consuming() 
 
消费者

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
import pika 
 
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.21.128')) 
channel = connection.channel() 
 
# make message persistent 
channel.queue_declare(queue='hello') 
 
 
def callback(ch, method, properties, body): 
    print(" [x] Received %r" % body) 
    import time 
    time.sleep(10) 
    print 'ok' 
    ch.basic_ack(delivery_tag = method.delivery_tag) 
 
channel.basic_qos(prefetch_count=1) 
 
channel.basic_consume(callback, 
                      queue='hello', 
                      no_ack=False) 
 
print(' [*] Waiting for messages. To exit press CTRL+C') 
channel.start_consuming() 
 
消费者

#!/usr/bin/env python 
import pika 
import sys 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
channel = connection.channel() 
 
channel.exchange_declare(exchange='logs', 
                         type='fanout') 
 
message = ' '.join(sys.argv[1:]) or "info: Hello World!" 
channel.basic_publish(exchange='logs', 
                      routing_key='', 
                      body=message) 
print(" [x] Sent %r" % message) 
connection.close()

#!/usr/bin/env python 
import pika 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
channel = connection.channel() 
 
channel.exchange_declare(exchange='logs', 
                         type='fanout') 
 
result = channel.queue_declare(exclusive=True) 
queue_name = result.method.queue 
 
channel.queue_bind(exchange='logs', 
                   queue=queue_name) 
 
print(' [*] Waiting for logs. To exit press CTRL+C') 
 
def callback(ch, method, properties, body): 
    print(" [x] %r" % body) 
 
channel.basic_consume(callback, 
                      queue=queue_name, 
                      no_ack=True) 
 
channel.start_consuming()

#!/usr/bin/env python 
import pika 
import sys 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
channel = connection.channel() 
 
channel.exchange_declare(exchange='direct_logs', 
                         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) 
 
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()

#!/usr/bin/env python 
import pika 
import sys 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
channel = connection.channel() 
 
channel.exchange_declare(exchange='direct_logs', 
                         type='direct') 
 
severity = sys.argv[1] if len(sys.argv) > 1 else 'info' 
message = ' '.join(sys.argv[2:]) or 'Hello World!' 
channel.basic_publish(exchange='direct_logs', 
                      routing_key=severity, 
                      body=message) 
print(" [x] Sent %r:%r" % (severity, message)) 
connection.close()

#!/usr/bin/env python 
import pika 
import sys 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
channel = connection.channel() 
 
channel.exchange_declare(exchange='topic_logs', 
                         type='topic') 
 
result = channel.queue_declare(exclusive=True) 
queue_name = result.method.queue 
 
binding_keys = sys.argv[1:] 
if not binding_keys: 
    sys.stderr.write("Usage: %s [binding_key].../n" % sys.argv[0]) 
    sys.exit(1) 
 
for binding_key in binding_keys: 
    channel.queue_bind(exchange='topic_logs', 
                       queue=queue_name, 
                       routing_key=binding_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()

#!/usr/bin/env python 
import pika 
import sys 
 
connection = pika.BlockingConnection(pika.ConnectionParameters( 
        host='localhost')) 
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()

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
 
from sqlalchemy import create_engine 
 
 
engine = create_engine("mysql+mysqldb://root:[email protected]:3306/s11", max_overflow=5) 
 
 
# 事务操作 
with engine.begin() as conn: 
    conn.execute("insert into table (x, y, z) values (1, 2, 3)") 
    conn.execute("my_special_procedure(5)") 
     
     
conn = engine.connect() 
# 事务操作  
with conn.begin(): 
       conn.execute("some statement", {'x':5, 'y':10})

#!/usr/bin/env python 
# -*- coding:utf-8 -*- 
 
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey 
 
metadata = MetaData() 
 
user = Table('user', metadata, 
    Column('id', Integer, primary_key=True), 
    Column('name', String(20)), 
) 
 
color = Table('color', metadata, 
    Column('id', Integer, primary_key=True), 
    Column('name', String(20)), 
) 
engine = create_engine("mysql+mysqldb://root:[email protected]:3306/s11", max_overflow=5) 
 
conn = engine.connect() 
 
# 创建SQL语句，INSERT INTO "user" (id, name) VALUES (:id, :name) 
conn.execute(user.insert(),{'id':7,'name':'seven'}) 
conn.close() 
 
# sql = user.insert().values(id=123, name='wu') 
# conn.execute(sql) 
# conn.close() 
 
# sql = user.delete().where(user.c.id > 1) 
 
# sql = user.update().values(fullname=user.c.name) 
# sql = user.update().where(user.c.name == 'jack').values(name='ed') 
 
# sql = select([user, ]) 
# sql = select([user.c.id, ]) 
# sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id) 
# sql = select([user.c.name]).order_by(user.c.name) 
# sql = select([user]).group_by(user.c.name) 
 
# result = conn.execute(sql) 
# print result.fetchall() 
# conn.close()