#!/usr/bin/env  python

# --*--coding:utf- --*--

import memcache

mc = memcache.Client(['192.168.17.129:12000'], debug=True)

mc.set_multi({'k6': 'v4', 'k7': 'v5'})

ret = mc.get('k6')

set = mc.get('k7')

print ret,set

class ConnectionPool(object):

    ...........

  def __init__(self, connection_class=Connection, max_connections=None,

         **connection_kwargs):  #类初始化时调用构造函数

    max_connections = max_connections or  **

    if not isinstance(max_connections, (int, long)) or max_connections < : #判断输入的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 =  #已经创建的连接的计数器

    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 +=   #把代表已经创建的连接的数值+

    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- -*-

import redis

class RedisHelper:

    def __init__(self):

        self.__conn = redis.Redis(host='192.168.17.128')#连接redis服务器

        self.chan_sub = 'fm104.5'#定义值

        self.chan_pub = 'fm104.5'#定义值

    def public(self, 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- -*-

import Queue

import threading

message = Queue.Queue()

def producer(i):

    while True:

        message.put(i)

def consumer(i):

    while True:

        msg = message.get()

for i in range():

    t = threading.Thread(target=producer, args=(i,))

    t.start()

for i in range():

    t = threading.Thread(target=consumer, args=(i,))

    t.start()

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='10.211.55.4'))

channel = connection.channel()

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):

    print(" [x] Received %r" % body)

    import time

    time.sleep()

    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='10.211.55.4'))

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=, # make message persistent

                      ))

print(" [x] Sent 'Hello World!'")

connection.close()

复制代码

#!/usr/bin/env python

# -*- coding:utf- -*-

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))

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()

    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- -*-

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))

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()

    print 'ok'

    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=)

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[:]) 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[:]

if not severities:

    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[])

    sys.exit()

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[] if len(sys.argv) >  else 'info'

message = ' '.join(sys.argv[:]) 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[:]

if not binding_keys:

    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[])

    sys.exit()

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[] if len(sys.argv) >  else 'anonymous.info'

message = ' '.join(sys.argv[:]) 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- -*-

from sqlalchemy import create_engine

engine = create_engine("mysql+mysqldb://root:123456@192.168.17.129:3306/wulaoer", max_overflow=)

# 事务操作

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':, 'y':})

#!/usr/bin/env python

# -*- coding:utf- -*-

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()),

)

color = Table('color', metadata,#创建列表

    Column('id', Integer, primary_key=True),

    Column('name', String()),

)

engine = create_engine("mysql+mysqldb://root:123456@192.168.17.129:3306/wulaoer", max_overflow=)

conn = engine.connect()

# 创建SQL语句，INSERT INTO "user" (id, name) VALUES (:id, :name)#增加数据

# conn.execute(user.insert(),{'id':,'name':'seven'})

# conn.close()

# sql = user.insert().values(id=, name='wu')#插入数据

# conn.execute(sql)

# conn.close()

sql = user.delete().where(user.c.id > )#删

# 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()