|
|
Memcached
Memcached 是一个高性能的分布式内存对象缓存系统，用于动态Web应用以减轻数据库负载。它通过在内存中缓存数据和对象来减少读取数据库的次数，从而提高动态、数据库驱动网站的速度。Memcached基于一个存储键/值对的hashmap。其守护进程（daemon ）是用C写的，但是客户端可以用任何语言来编写，并通过memcached协议与守护进程通信。
Python操作Memcached
安装API
python操作Memcached使用Python-memcached模块
下载安装：https://pypi.python.org/pypi/python-memcached
连接操作
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set("foo", "bar")
ret = mc.get('foo')
print ret
支持集群
python-memcached模块原生支持集群操作，其原理是在内存维护一个主机列表，且集群中主机的权重值和主机在列表中重复出现的次数成正比
主机 权重
1.1.1.1 1
1.1.1.2 2
1.1.1.3 1
那么在内存中主机列表为：
host_list = ["1.1.1.1", "1.1.1.2", "1.1.1.2", "1.1.1.3", ]
如果用户根据如果要在内存中创建一个键值对（如：k1 = "v1"），那么要执行一下步骤：
- 根据算法将 k1 转换成一个数字
- 将数字和主机列表长度求余数，得到一个值 N（ 0 <= N < 列表长度 ）
- 在主机列表中根据 第2步得到的值为索引获取主机，例如：host_list[N]
- 连接 将第3步中获取的主机，将 k1 = "v1" 放置在该服务器的内存中
add
添加一条键值对，如果已经存在的 key，重复执行add操作异常
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.add('k1', 'v1')
# mc.add('k1', 'v2') # 报错，对已经存在的key重复添加，失败！！！
replace
replace 修改某个key的值，如果key不存在，则异常
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
# 如果memcache中存在kkkk，则替换成功，否则一场
mc.replace('kkkk','999')
set 和set_multi
set 设置一个键值对，如果key不存在，则创建，如果key存在，则修改
set_multi 设置多个键值对，如果key不存在，则创建，如果key存在，则修改
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set('key0', 'wupeiqi')
mc.set_multi({'key1': 'val1', 'key2': 'val2'})
delete 和delete_multi
delete 在Memcached中删除指定的一个键值对
delete_multi 在Memcached中删除指定的多个键值对
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.delete('key0')
mc.delete_multi(['key1', 'key2'])
get 和get_multi
get 获取一个键值对
get_multi 获取多一个键值对
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
val = mc.get('key0')
item_dict = mc.get_multi(["key1", "key2", "key3"])
append 和prepend
append 修改指定key的值，在该值 后面 追加内容
prepend 修改指定key的值，在该值 前面 插入内容
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
# k1 = "v1"
mc.append('k1', 'after')
# k1 = "v1after"
mc.prepend('k1', 'before')
# k1 = "beforev1after"
decr 和incr
incr 自增，将Memcached中的某一个值增加 N （ N默认为1 ）
decr 自减，将Memcached中的某一个值减少 N （ N默认为1 ）
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set('k1', '777')
mc.incr('k1')
# k1 = 778
mc.incr('k1', 10)
# k1 = 788
mc.decr('k1')
# k1 = 787
mc.decr('k1', 10)
# k1 = 777
gets 和cas
如商城商品剩余个数，假设改值保存在memcache中，product_count = 900
A用户刷新页面从memcache中读取到product_count = 900
B用户刷新页面从memcache中读取到product_count = 900
如果A、B用户均购买商品
A用户修改商品剩余个数 product_count＝899
B用户修改商品剩余个数 product_count＝899
如此一来缓存内的数据便不在正确，两个用户购买商品后，商品剩余还是 899
如果使用python的set和get来操作以上过程，那么程序就会如上述所示情况！
如果想要避免此情况的发生，只要使用 gets 和 cas 即可，如：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True, cache_cas=True)
v = mc.gets('product_count')
# ...
# 如果有人在gets之后和cas之前修改了product_count，那么，下面的设置将会执行失败，剖出异常，从而避免非正常数据的产生
mc.cas('product_count', "899")
RabbitMQ
RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
安装API
pip install pika
对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。
#!/usr/bin/env python
import pika
# ######################### 生产者 #########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
#!/usr/bin/env python
import pika
# ########################## 消费者 ##########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(callback,
queue='hello',
no_ack=True)
print('
Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
durable 消息不丢失
#生产者
#!/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=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='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(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()
消息获取顺序
默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者1去队列中获取 偶数 序列的任务。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
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(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()
发布订阅
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
exchange type = fanout
#生产者
#!/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()
关键字发送
exchange type = direct
之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
#消费者
#!/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()
模糊匹配
exchange type = topic
在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。
- # 表示可以匹配 0 个 或 多个 单词
- * 表示只能匹配 一个 单词
发送者路由值 队列中
old.boy.python old.* -- 不匹配
old.boy.python old.# -- 匹配
#消费者
#!/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()
RabbitMQ rpc模式
#server ， 生产者
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import uuid
import json
conn = pika.BlockingConnection(pika.ConnectionParameters(host='10.37.129.5'))
channel = conn.channel()
def callback(ch, method, properties, body):
print(body)
ch.queue_delete(method.routing_key)
def send_msg(hostname, cmd):
"""
向队列中发送命令，并等待命令在客户端执行完成后获取结果
:param hostname:
:param cmd:
:return:
"""
# 创建临时队列，用于存放客户端执行命令后的返回值
queue_name = str(uuid.uuid4())
channel.queue_declare(queue=queue_name)
# 向客户端队列中发送命令：封装了命令以及执行结果存放的队列名称
body = {'uuid': queue_name, 'content': cmd}
channel.basic_publish(exchange='', routing_key=hostname, body=json.dumps(body))
# 等待客户端想队列中发送执行结果，超时时间10s
v = channel.consume(queue_name, inactivity_timeout=10)
try:
for method, properties, body in v:
# 执行指定回调函数
callback(channel, method, properties, body)
except TypeError as e:
# 如果超时，则删除临时队列，不再获取数据
channel.queue_delete(queue_name)
if __name__ == '__main__':
while True:
hostname = input('hostname( c1.com 或 c2.com ):')
cmd = input('cmd:')
if cmd == 'exit':
break
send_msg(hostname, cmd)
conn.close()
#agent 消费者
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import json
import subprocess
conn = pika.BlockingConnection(pika.ConnectionParameters(host='10.37.129.5'))
channel = conn.channel()
channel.queue_declare(queue='c2.com')
def callback(ch, method, properties, body):
body = json.loads(str(body, encoding='utf-8'))
result = subprocess.getoutput(body['content'])
result = 'c2.com:%s' % result
ch.basic_publish(exchange='', routing_key=body['uuid'], body=result)
channel.basic_consume(callback, queue='c2.com', no_ack=True)
channel.start_consuming()
程序练习：
基于主机管理的程序把ssh换成rpc的连接方式
github：https://github.com/wangyufu/host_manage_rpc |
|
QQ群⑧：
窥视卡
雷达卡
发表于 2017-7-2 17:57:52
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡