1.线程池

1.1 contextlib ，上下文切换

import contextlib

#这个装饰器赋予work_state这个方法能使用“with work_state”，实现了一个上下文切换的功能
@contextlib.contextmanager
def work_state(state_list,work_thread):  #①
    state_list.append(work_thread)           #②
    try:
        yield                                               #③
    finally:
        state_list.remove(work_thread)       #⑨

state_list = []
work_thread = 'fdafd'

with work_state(state_list,work_thread): #④
    print(123)                                           #⑤
    print(state_list)                                   #⑥
    print(work_thread)                              #⑦
    print(456)                                           #⑧

#结果：
123
['fdafd']
fdafd
456

import contextlib
import socket

@contextlib.contextmanager
def context_sock(host,port):
    sk = socket.socket()
    sk.bind((host,port))
    sk.listen(5)
    try:
        yield sk    #将sk传给下面的“with context_sock...as sock”
    finally:
        sk.close()
with context_sock('localhost',8888) as sock:
    print(sock)   #此处的sock就是sk。

2.redis 补充

redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
import redis
 
#创建一个连接池，之后与redis的操作就可以通过调用这个连接池来进行。
pool = redis.ConnectionPool(host='10.211.55.4', port=6379)
 
r = redis.Redis(connection_pool=pool)
r.set('foo', 'Bar')
print(r.get('foo'))

2.1redis发布订阅

#redis_pub_sub.py
import redis
class RedisHelper(object):
    def __init__(self):
        self.__conn = redis.Redis(host='127.0.0.1')

    def public(self,msg,channel):
        self.__conn.publish(channel,msg)   #先channel后msg。
        return True

    def subscribe(self,channel):
        pub = self.__conn.pubsub()
        pub.subscribe(channel)
        pub.parse_response()
        return pub

#sub.py
import redis_pub_sub

sub1 = redis_pub_sub.RedisHelper()
data = sub1.subscribe('fm104.99')
print(data.parse_response())
#可以起多个sub，当pub发布了消息，这些sub都能收到。

#pub.py
import redis_pub_sub

pub1 = redis_pub_sub.RedisHelper()
pub1.public('avadv','fm104.99')

3. 初识rabbitMQ

rabbitmq实现发布订阅；pika是使用rabbitmq的插件。

#publisher

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='first')
print('publish ok')
connection.close()


#consumer
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()
channel.queue_declare(queue='hello')

def callable(ch,method,properties,body):
    print("[x] Received %r" % body)

channel.basic_consume(callable,
                      queue='hello',
                      no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

3.1 ackownledgment，消息不丢失

上面的代码实现了发布订阅，但是有个问题，假如consumer在接收消息的过程中宕了，就会出现consumer并没有接受完消息，但是消息队列里也没消息了，即使consumer恢复了也没法再重新获取到消息了，这时候就需要设置一下消息不丢失。

消息不丢失只需要修改consumer端的代码即可，其实是改成“consumer接受完消息后，会给消息队列一个接收完毕的消息，然后这条消息就从队列里移除了，如果consumer没有发确认消息，此条消息就一直存在消息队列里”。

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()
channel.queue_declare(queue='hello')

def callable(ch,method,properties,body):
    print("[x] Received %r" % body)
    import time
    time.sleep(50)
    print('consumer ok')
    ch.basic_ack(delivery_tag = method.delivery_tag) #接收完消息后，给消息队列发确认收到的消息。

channel.basic_consume(callable,
                      queue='hello',
                      no_ack=False      #这一行表示需要消息确认；no_ack的全称是no ackownledge。
                      )
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

3.2 during,消息持久化

消息持久化是在publish端配置

#!/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, # 设置了这里，就使消息持久化了。
                      ))
print(" [x] Sent 'Hello World!'")
connection.close()

publish端配置了持久化，consumer端也配置了持久化，这样无论哪端宕了都不会丢失消息。

3.3 消息获取顺序

默认的话，各个consumer获取消息的顺序是轮询来取的，consumer1取消息1，consumer2取消息2，consumer3取消息3，这时候如果consumer3先处理完了，需要再取消息，那consumer3取的是6，因为4和5之后会被consumer1和consumer2取走。

如果想做到哪个consumer先处理完了就接着取接下来的消息的话，即各个consumer顺序取消息，那就需要在consumer代码里加下述参数：channel.basic_qos(prefetch_count=1)

这样如果consumer3先处理完了，再取消息就是消息4了，而不是消息6。

3.4 发布和订阅

3.4.1 exchange

exchange是publisher和rabbitmq间的一个中间层，publisher给rabbitmq发消息，其实是先发给了exchange，然后exchange根据规则发到对应的消息队列里。

exchange的作用，比如你想给所有的消息队列发一条消息，此时rabbitmq里有100个消息队列，如果没有exchange的话，你需要自己发送100次消息，如果有了exchange，你需要做的仅仅是将消息发送给rabbitmq，然后选择一个对应的规则，比如说是广播，那exchange就会把消息发给所有绑定到exchange的队列。这里仅仅是举了一个广播的例子，exchange还有其他发布模式。

3.4.2 exchange的fanout模式，广播

publisher

#!/usr/bin/env python


import pika

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

channel = connection.channel()

#声明一个exchange，名称和类型
channel.exchange_declare(exchange='logs_exchange' , type='fanout')

message = 'hhh'

#向这个exchange发送消息
channel.basic_publish(exchange='logs_exchange',routing_key='',body=message)


print('publish ok')

connection.close()

consumer

#!/usr/bin/env python

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

#声明一个exchange
channel.exchange_declare(exchange='logs_exchange',type='fanout')

#下面两行代码的作用是声明一个随机名字的消息队列，这样每次执行这些代码，就会生成不同的消息队列；如果指定了队列名字，那每次执行都是同一个队列。
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

#绑定exchange
channel.queue_bind(exchange='logs_exchange',queue=queue_name)

print('waiting the message coming...')

def callable(ch,method,properties,body):
    print('recieve the %s' % body)

channel.basic_consume(callable,queue=queue_name,no_ack=True)

channel.start_consuming()

多次执行consumer的代码，生成了多个消息队列，所有的consumer都会同时收到publisher的消息。

3.4.3 exchange的direct模式，基于关键字来发送消息

发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。

publisher指定将消息发给绑定了info这个关键字的队列；consumer1绑定了info、error关键字；consumer2绑定了error关键字；则，只有consumer1能收到publisher的消息。如果publisher将消息发给绑定了error关键字的队列，则consumer1和consumer2都能收到消息。这就是基于关键字来管理消息。

生产者

#!/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='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]
" % sys.argv[0])
    sys.exit(1)

#循环绑定关键字；routing_key是变化的
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()

3.4.4 模糊匹配

在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

# 表示可以匹配 0 个或多个单词
* 表示只能匹配一个单词

发送者路由值              队列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配

saltstack支持模糊匹配，虽然saltstack用的zeromq，但是模糊匹配的原理跟这个类似。

生产者

#!/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
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]...
" % 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()