Day9 - 进程、线程、协程篇（二）

时间  2019-11-25
标签 day9 day 进程 线程 繁體版
原文   原文链接

本节内容

  1. Gevent协程
  2. Select\Poll\Epoll异步IO与事件驱动
  3. Python链接Mysql数据库操做
  4. RabbitMQ队列
  5. Redis\Memcached缓存
  6. Paramiko SSH
  7. Twsited网络框架

 

一、协程（单线程下实现多并发的效果）

协程，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程。（CPU不认识协程，是用户本身操做的）html

协程拥有本身的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其余地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。所以：python

协程能保留上一次调用时的状态（即全部局部状态的一个特定组合），每次过程重入时，就至关于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。mysql

 

协程的好处：react

  • 无需线程上下文切换的开销（单线程，函数的来回切换，yield）
    • 无需原子操做锁定及同步的开销（单线程，不涉及多线程同时操做一个内存变量，即同时改数据）
    • "原子操做(atomic operation)是不须要synchronized"，所谓原子操做是指不会被线程调度机制打断的操做；这种操做一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另外一个线程）。原子操做能够是一个步骤，也能够是多个操做步骤，可是其顺序是不能够被打乱，或者切割掉只执行部分。视做总体是原子性的核心。
  • 方便切换控制流，简化编程模型
  • 高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。因此很适合用于高并发处理。

 

缺点：git

  • 没法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程须要和进程配合才能运行在多CPU上.固然咱们平常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
  • 进行阻塞（Blocking）操做（如IO时）会阻塞掉整个程序

使用yield实现协程操做例子　　　　程序员

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import  time
import  queue
def  consumer(name):
     print ( "--->starting eating baozi..." )
     while  True :
         new_baozi  =  yield
         print ( "[%s] is eating baozi %s"  %  (name,new_baozi))
         #time.sleep(1)
 
def  producer():
 
     r  =  con.__next__()
     r  =  con2.__next__()
     n  =  0
     while  n <  5 :
         n  + = 1
         con.send(n)
         con2.send(n)
         print ( "\033[32;1m[producer]\033[0m is making baozi %s"  % n )
 
 
if  __name__  = =  '__main__' :
     con  =  consumer( "c1" )
     con2  =  consumer( "c2" )
     p  =  producer()

看楼上的例子，我问你这算不算作是协程呢？你说，我他妈哪知道呀，你前面说了一堆废话，可是并没告诉我协程的标准形态呀，我腚眼一想，以为你说也对，那好，咱们先给协程一个标准定义，即符合什么条件就能称之为协程：github

  1. 必须在只有一个单线程里实现并发
  2. 修改共享数据不需加锁
  3. 用户程序里本身保存多个控制流的上下文栈
  4. 一个协程遇到IO操做自动切换到其它协程

基于上面这4点定义，咱们刚才用yield实现的程并不能算是合格的线程，由于它有一点功能没实现，哪一点呢？redis

 

Greenlet

greenlet是一个用C实现的协程模块，相比与python自带的yield，它可使你在任意函数之间随意切换，而不需把这个函数先声明为generatorsql

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# -*- coding:utf-8 -*-
 
 
from  greenlet  import  greenlet
 
 
def  test1():
     print ( 12 )
     gr2.switch()
     print ( 34 )
     gr2.switch()
 
 
def  test2():
     print ( 56 )
     gr1.switch()
     print ( 78 )
 
 
gr1  =  greenlet(test1)
gr2  =  greenlet(test2)
gr1.switch()

感受确实用着比generator还简单了呢，但好像尚未解决一个问题，就是遇到IO操做，自动切换，对不对？数据库

 

　　

 

　　

Gevent 

Gevent 是一个第三方库，能够轻松经过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet所有运行在主程序操做系统进程的内部，但它们被协做式地调度。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import  gevent
 
def  func1():
     print ( '\033[31;1m李闯在跟海涛搞...\033[0m' )
     gevent.sleep( 2 )
     print ( '\033[31;1m李闯又回去跟继续跟海涛搞...\033[0m' )
 
def  func2():
     print ( '\033[32;1m李闯切换到了跟海龙搞...\033[0m' )
     gevent.sleep( 1 )
     print ( '\033[32;1m李闯搞完了海涛，回来继续跟海龙搞...\033[0m' )
 
 
gevent.joinall([
     gevent.spawn(func1),
     gevent.spawn(func2),
     #gevent.spawn(func3),
])

 

同步与异步的性能区别 

import gevent
 
def task(pid):
    """
    Some non-deterministic task
    """
    gevent.sleep(0.5)
    print('Task %s done' % pid)
 
def synchronous():
    for i in range(1,10):
        task(i)
 
def asynchronous():
    threads = [gevent.spawn(task, i) for i in range(10)]
    gevent.joinall(threads)
 
print('Synchronous:')
synchronous()
 
print('Asynchronous:')
asynchronous()
View Code

上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 初始化的greenlet列表存放在数组threads中，此数组被传给gevent.joinall 函数，后者阻塞当前流程，并执行全部给定的greenlet。执行流程只会在 全部greenlet执行完后才会继续向下走。　　

遇到IO阻塞时会自动切换任务

from gevent import monkey; monkey.patch_all()
import gevent
from  urllib.request import urlopen
 
def f(url):
    print('GET: %s' % url)
    resp = urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))
 
gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])
View Code

 

经过gevent实现单线程下的多socket并发

import sys
import socket
import time
import gevent
 
from gevent import socket,monkey
monkey.patch_all()
 
 
def server(port):
    s = socket.socket()
    s.bind(('0.0.0.0', port))
    s.listen(500)
    while True:
        cli, addr = s.accept()
        gevent.spawn(handle_request, cli)
 
 
 
def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            print("recv:", data)
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)
 
    except Exception as  ex:
        print(ex)
    finally:
        conn.close()
if __name__ == '__main__':
    server(8001)
server side  
import socket
 
HOST = 'localhost'    # The remote host
PORT = 8001           # The same port as used by the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
    msg = bytes(input(">>:"),encoding="utf8")
    s.sendall(msg)
    data = s.recv(1024)
    #print(data)
 
    print('Received', repr(data))
s.close()
client side 
import socket
import threading

def sock_conn():

    client = socket.socket()

    client.connect(("localhost",8001))
    count = 0
    while True:
        #msg = input(">>:").strip()
        #if len(msg) == 0:continue
        client.send( ("hello %s" %count).encode("utf-8"))

        data = client.recv(1024)

        print("[%s]recv from server:" % threading.get_ident(),data.decode()) #结果
        count +=1
    client.close()


for i in range(100):
    t = threading.Thread(target=sock_conn)
    t.start()
并发100个sock链接

 

　　

论事件驱动与异步IO

一般，咱们写服务器处理模型的程序时，有如下几种模型：
（1）每收到一个请求，建立一个新的进程，来处理该请求；
（2）每收到一个请求，建立一个新的线程，来处理该请求；
（3）每收到一个请求，放入一个事件列表，让主进程经过非阻塞I/O方式来处理请求
上面的几种方式，各有千秋，
第（1）中方法，因为建立新的进程的开销比较大，因此，会致使服务器性能比较差,但实现比较简单。
第（2）种方式，因为要涉及到线程的同步，有可能会面临 死锁等问题。
第（3）种方式，在写应用程序代码时，逻辑比前面两种都复杂。
综合考虑各方面因素，通常广泛认为第（3）种方式是大多数 网络服务器采用的方式
 

看图说话讲事件驱动模型

在UI编程中，经常要对鼠标点击进行相应，首先如何得到鼠标点击呢？
方式一：建立一个线程，该线程一直循环检测是否有鼠标点击，那么这个方式有如下几个缺点：
1. CPU资源浪费，可能鼠标点击的频率很是小，可是扫描线程仍是会一直循环检测，这会形成不少的CPU资源浪费；若是扫描鼠标点击的接口是阻塞的呢？
2. 若是是堵塞的，又会出现下面这样的问题，若是咱们不但要扫描鼠标点击，还要扫描键盘是否按下，因为扫描鼠标时被堵塞了，那么可能永远不会去扫描键盘；
3. 若是一个循环须要扫描的设备很是多，这又会引来响应时间的问题；
因此，该方式是很是很差的。

方式二：就是事件驱动模型
目前大部分的UI编程都是事件驱动模型，如不少UI平台都会提供onClick()事件，这个事件就表明鼠标按下事件。事件驱动模型大致思路以下：
1. 有一个事件（消息）队列；
2. 鼠标按下时，往这个队列中增长一个点击事件（消息）；
3. 有个循环，不断从队列取出事件，根据不一样的事件，调用不一样的函数，如onClick()、onKeyDown()等；
4. 事件（消息）通常都各自保存各自的处理函数指针，这样，每一个消息都有独立的处理函数；

 

 

 

事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。它的特色是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是（单线程）同步以及多线程编程。

让咱们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展现了随着时间的推移，这三种模式下程序所作的工做。这个程序有3个任务须要完成，每一个任务都在等待I/O操做时阻塞自身。阻塞在I/O操做上所花费的时间已经用灰色框标示出来了。

 

在单线程同步模型中，任务按照顺序执行。若是某个任务由于I/O而阻塞，其余全部的任务都必须等待，直到它完成以后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。若是任务之间并无互相依赖的关系，但仍然须要互相等待的话这就使得程序没必要要的下降了运行速度。

在多线程版本中，这3个任务分别在独立的线程中执行。这些线程由操做系统来管理，在多处理器系统上能够并行处理，或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其余线程得以继续执行。与完成相似功能的同步程序相比，这种方式更有效率，但程序员必须写代码来保护共享资源，防止其被多个线程同时访问。多线程程序更加难以推断，由于这类程序不得不经过线程同步机制如锁、可重入函数、线程局部存储或者其余机制来处理线程安全问题，若是实现不当就会致使出现微妙且使人痛不欲生的bug。

在事件驱动版本的程序中，3个任务交错执行，但仍然在一个单独的线程控制中。当处理I/O或者其余昂贵的操做时，注册一个回调到事件循环中，而后当I/O操做完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询全部的事件，当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽量的得以执行而不须要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为，由于程序员不须要关心线程安全问题。

当咱们面对以下的环境时，事件驱动模型一般是一个好的选择：

  1. 程序中有许多任务，并且…
  2. 任务之间高度独立（所以它们不须要互相通讯，或者等待彼此）并且…
  3. 在等待事件到来时，某些任务会阻塞。

当应用程序须要在任务间共享可变的数据时，这也是一个不错的选择，由于这里不须要采用同步处理。

网络应用程序一般都有上述这些特色，这使得它们可以很好的契合事件驱动编程模型。

 

此处要提出一个问题，就是，上面的事件驱动模型中，只要一遇到IO就注册一个事件，而后主程序就能够继续干其它的事情了，只到io处理完毕后，继续恢复以前中断的任务，这本质上是怎么实现的呢？哈哈，下面咱们就来一块儿揭开这神秘的面纱。。。。

 

 

二、Select\Poll\Epoll异步IO　

http://www.cnblogs.com/alex3714/p/4372426.html　

番外篇 http://www.cnblogs.com/alex3714/articles/5876749.html 

select 多并发socket 例子

  select socket server
  select socket client

 

 

（2）selectors模块（I/O复用）

适合应用在FTP的server端，支持多链接并发上传下载文件

  1 """
  2 SELECT版超级简单版服务器端
  3 目前只实现了上传，参考read()函数
  4 使用方法：直接执行便可
  5 待优化：未对断线的客户端进行处理，好比断线后fp未关闭
  6 """
  7 import os, sys, datetime, time
  8 import selectors
  9 import socket
 10 
 11 sel = selectors.SelectSelector()
 12 log_flag = True
 13 
 14 # accept函数直接拷贝官方文档的，不是重点
 15 def accept(sock, mask):
 16     conn, addr = sock.accept()  # Should be ready
 17     print('accepted', conn, 'from', addr)
 18     conn.setblocking(False)#设置为不阻塞
 19     #注册监听事件：read
 20     sel.register(conn, selectors.EVENT_READ, read)
 21 
 22 def log(msg):
 23     if log_flag:
 24         with open("log.txt", "a", encoding="utf8") as f:
 25             f.write(msg + "\n")
 26 
 27 # 关键代码↓
 28 upload_jobs = dict()   # 上传的任务，写成一个字典存放，以连接为key
 29 download_jobs = dict()   # 下载的任务，写成一个字典存放
 30 def read(conn, mask):
 31     try:
 32         data = conn.recv(4096)  # Should be ready
 33         if data:    # 有数据
 34             if data.startswith(b"put"):
 35                 # 当这个数据是一条上传指令，则进行如下初始化任务
 36                 filename = data.split(b"|")[1]   # 接受到的命令格式应该为put|filename|filesize
 37                 filesize = int(data.split(b"|")[2])
 38                 upload_jobs[conn] = dict(filename=filename, filesize=filesize, received_size=0, fp=open(filename, mode='ab'))
 39                 conn.send(b'1')
 40             elif data.startswith(b"get"):
 41                 # 下载指令，初始化任务
 42                 pass
 43             else:
 44                 if conn in upload_jobs:
 45                     # 若是本连接是上传任务
 46                     fp = upload_jobs[conn]['fp']
 47                     remain_size = upload_jobs[conn]['filesize'] - upload_jobs[conn]['received_size']    # 得到余下文件数据的长度
 48                     if remain_size <= 4096:  # 若是少于或等于4096个字节，表明这是最后一次读取
 49                         data = data[:remain_size]
 50                         fp.write(data)
 51                         fp.flush()
 52                         fp.close()
 53                         del upload_jobs[conn]   # 写入完毕后，从上传任务中移出该连接
 54                     else:
 55                         # 余下超过4096个字节，则把读取到的data所有写入
 56                         fp.write(data)
 57                         fp.flush()
 58                         upload_jobs[conn]['received_size'] += len(data) # 更新已接受的总大小
 59                     msg = "收到来自%s，数据长度为%d字节" % (str(conn), len(data))
 60                     log(msg)
 61                     print(msg)
 62                 elif conn in download_jobs:
 63                     # 若是本连接是下载任务
 64                     pass
 65                 else:
 66                     # 不知道这个连接是什么鬼，把收到的数据发还回去
 67                     conn.send(data)
 68 
 69         else:
 70             print('closing', conn)
 71             sel.unregister(conn)# 注销事件
 72             conn.close()
 73     except ConnectionResetError:
 74         print('closing', conn)
 75         sel.unregister(conn)# 注销事件
 76         conn.close()
 77 # 关键代码↑
 78 
 79 
 80 # 如下代码不是重点，稍微看一下便可
 81 host = 'localhost'
 82 port = 1234
 83 sock = socket.socket()
 84 sock.bind((host, port))
 85 sock.listen(100)
 86 sock.setblocking(False)#默认不阻塞
 87 print("开始侦听 %s:%d" % (host, port))
 88 #注册selector事件：accept
 89 sel.register(sock, selectors.EVENT_READ, accept)
 90 
 91 while True:
 92     events = sel.select() #默认阻塞，有活动链接就返回活动的链接列表
 93     for key, mask in events:
 94         callback = key.data #accept
 95         callback(key.fileobj, mask) #key.fileobj=  文件句柄
 96 """
 97 # 从新注册事件，以更换回调函数，进入发送文件大小
 98 Sel.unregister(conn) #注销事件
 99 Sel.register(conn,selectors.EVENT_READ,回调函数)
100 """
selector_ftp_server

 

数据库操做与Paramiko模块 

http://www.cnblogs.com/wupeiqi/articles/5095821.html 

 

 

RabbitMQ队列　　

安装 http://www.rabbitmq.com/install-standalone-mac.html

安装python rabbitMQ module 

1
2
3
4
5
6
7
pip install pika
or
easy_install pika
or
源码
  
https: / / pypi.python.org / pypi / pika

实现最简单的队列通讯

 

send端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#!/usr/bin/env python
import  pika
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
                'localhost' ))
channel  =  connection.channel()
 
#声明queue
channel.queue_declare(queue = 'hello' )
 
#n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange = '',
                       routing_key = 'hello' ,
                       body = 'Hello World!' )
print ( " [x] Sent 'Hello World!'" )
connection.close()

receive端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#_*_coding:utf-8_*_
__author__  =  'Alex Li'
import  pika
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
                'localhost' ))
channel  =  connection.channel()
 
 
#You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
#was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
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()

 

远程链接rabbitmq server的话，须要配置权限 噢 

首先在rabbitmq server上建立一个用户

1
sudo  rabbitmqctl  add_user alex alex3714　　

同时还要配置权限，容许从外面访问

1
sudo  rabbitmqctl set_permissions -p / alex  ".*"  ".*"  ".*"

set_permissions [-p vhost] {user} {conf} {write} {read}

vhost

The name of the virtual host to which to grant the user access, defaulting to /.

user

The name of the user to grant access to the specified virtual host.

conf

A regular expression matching resource names for which the user is granted configure permissions.

write

A regular expression matching resource names for which the user is granted write permissions.

read

A regular expression matching resource names for which the user is granted read permissions.

 

 

 

　　

客户端链接的时候须要配置认证参数

1
2
3
4
5
6
credentials  =  pika.PlainCredentials( 'alex' ,  'alex3714' )
 
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
     '10.211.55.5' , 5672 , '/' ,credentials))
channel  =  connection.channel()

　　

　　

Work Queues

在这种模式下，RabbitMQ会默认把p发的消息依次分发给各个消费者(c),跟负载均衡差很少

消息提供者代码

　　

 

消费者代码

　　

 

此时，先启动消息生产者，而后再分别启动3个消费者，经过生产者多发送几条消息，你会发现，这几条消息会被依次分配到各个消费者身上　　

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code once RabbitMQ delivers message to the customer it immediately removes it from memory. In this case, if you kill a worker we will lose the message it was just processing. We'll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don't want to lose any tasks. If a worker dies, we'd like the task to be delivered to another worker.

In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back from the consumer to tell RabbitMQ that a particular message had been received, processed and that RabbitMQ is free to delete it.

If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.

There aren't any message timeouts; RabbitMQ will redeliver the message when the consumer dies. It's fine even if processing a message takes a very, very long time.

Message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the no_ack=True flag. It's time to remove this flag and send a proper acknowledgment from the worker, once we're done with a task.

1
2
3
4
5
6
7
8
def  callback(ch, method, properties, body):
     print  " [x] Received %r"  %  (body,)
     time.sleep( body.count( '.' ) )
     print  " [x] Done"
     ch.basic_ack(delivery_tag  =  method.delivery_tag)
 
channel.basic_consume(callback,
                       queue = 'hello' )

　　Using this code we can be sure that even if you kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered

　　　　

消息持久化　　

We have learned how to make sure that even if the consumer dies, the task isn't lost(by default, if wanna disable  use no_ack=True). But our tasks will still be lost if RabbitMQ server stops.

When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren't lost: we need to mark both the queue and messages as durable.

First, we need to make sure that RabbitMQ will never lose our queue. In order to do so, we need to declare it as durable:

1
channel.queue_declare(queue = 'hello' , durable = True )

　　

Although this command is correct by itself, it won't work in our setup. That's because we've already defined a queue called hello which is not durable. RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that. But there is a quick workaround - let's declare a queue with different name, for exampletask_queue:

1
channel.queue_declare(queue = 'task_queue' , durable = True )

　　

This queue_declare change needs to be applied to both the producer and consumer code.

At that point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by supplying a delivery_mode property with a value 2.

1
2
3
4
5
6
channel.basic_publish(exchange = '',
                       routing_key = "task_queue" ,
                       body = message,
                       properties = pika.BasicProperties(
                          delivery_mode  =  2 ,  # make message persistent
                       ))

消息公平分发

若是Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，极可能出现，一个机器配置不高的消费者那里堆积了不少消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，能够在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。

 

1
channel.basic_qos(prefetch_count = 1 )

 

带消息持久化+公平分发的完整代码

生产者端

消费者端

　　

Publish\Subscribe(消息发布\订阅)　

以前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被全部的Queue收到，相似广播的效果，这时候就要用到exchange了，

An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.

Exchange在定义的时候是有类型的，以决定究竟是哪些Queue符合条件，能够接收消息


fanout: 全部bind到此exchange的queue均可以接收消息
direct: 经过routingKey和exchange决定的那个惟一的queue能够接收消息
topic:全部符合routingKey(此时能够是一个表达式)的routingKey所bind的queue能够接收消息

　　 表达式符号说明：#表明一个或多个字符，*表明任何字符
      例：#.a会匹配a.a，aa.a，aaa.a等
          *.a会匹配a.a，b.a，c.a等
     注：使用RoutingKey为#，Exchange Type为topic的时候至关于使用fanout　

headers: 经过headers 来决定把消息发给哪些queue

消息publisher

消息subscriber

　　

有选择的接收消息(exchange type=direct)　

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 断定应该将数据发送至指定队列。

publisher

subscriber　

　　

更细致的消息过滤

Although using the direct exchange improved our system, it still has limitations - it can't do routing based on multiple criteria.

In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).

That would give us a lot of flexibility - we may want to listen to just critical errors coming from 'cron' but also all logs from 'kern'.

publisher

subscriber

To receive all the logs run:

python receive_logs_topic.py "#"

To receive all logs from the facility "kern":

python receive_logs_topic.py "kern.*"

Or if you want to hear only about "critical" logs:

python receive_logs_topic.py "*.critical"

You can create multiple bindings:

python receive_logs_topic.py "kern.*" "*.critical"

And to emit a log with a routing key "kern.critical" type:

python emit_log_topic.py "kern.critical" "A critical kernel error"

　　

Remote procedure call (RPC)

To illustrate how an RPC service could be used we're going to create a simple client class. It's going to expose a method named call which sends an RPC request and blocks until the answer is received:

1
2
3
fibonacci_rpc  =  FibonacciRpcClient()
result  =  fibonacci_rpc.call( 4 )
print ( "fib(4) is %r"  %  result)

RPC server

RPC client

　　

　　

Memcached & Redis使用　

memcached 

http://www.cnblogs.com/wupeiqi/articles/5132791.html 　

 

redis 使用

http://www.cnblogs.com/alex3714/articles/6217453.html　　

　

Twsited异步网络框架

Twisted是一个事件驱动的网络框架，其中包含了诸多功能，例如：网络协议、线程、数据库管理、网络操做、电子邮件等。　

事件驱动

简而言之，事件驱动分为二个部分：第一，注册事件；第二，触发事件。

自定义事件驱动框架，命名为：“弑君者”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
# event_drive.py
 
event_list  =  []
 
 
def  run():
     for  event  in  event_list:
         obj  =  event()
         obj.execute()
 
 
class  BaseHandler( object ):
     """
     用户必须继承该类，从而规范全部类的方法（相似于接口的功能）
     """
     def  execute( self ):
         raise  Exception( 'you must overwrite execute' )
 
最牛逼的事件驱动框架

程序员使用“弑君者框架”：　　

1
2
3
4
5
6
7
8
9
10
11
12
13
14
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
from  source  import  event_drive
 
 
class  MyHandler(event_drive.BaseHandler):
 
     def  execute( self ):
         print  'event-drive execute MyHandler'
 
 
event_drive.event_list.append(MyHandler)
event_drive.run()

 

Protocols

Protocols描述了如何以异步的方式处理网络中的事件。HTTP、DNS以及IMAP是应用层协议中的例子。Protocols实现了IProtocol接口，它包含以下的方法：

makeConnection 在transport对象和服务器之间创建一条链接 connectionMade 链接创建起来后调用 dataReceived 接收数据时调用 connectionLost 关闭链接时调用

Transports

Transports表明网络中两个通讯结点之间的链接。Transports负责描述链接的细节，好比链接是面向流式的仍是面向数据报的，流控以及可靠性。TCP、UDP和Unix套接字可做为transports的例子。它们被设计为“知足最小功能单元，同时具备最大程度的可复用性”，并且从协议实现中分离出来，这让许多协议能够采用相同类型的传输。Transports实现了ITransports接口，它包含以下的方法：

write 以非阻塞的方式按顺序依次将数据写到物理链接上 writeSequence 将一个字符串列表写到物理链接上 loseConnection 将全部挂起的数据写入，而后关闭链接 getPeer 取得链接中对端的地址信息 getHost 取得链接中本端的地址信息

将transports从协议中分离出来也使得对这两个层次的测试变得更加简单。能够经过简单地写入一个字符串来模拟传输，用这种方式来检查。

　　

 

EchoServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from  twisted.internet  import  protocol
from  twisted.internet  import  reactor
 
class  Echo(protocol.Protocol):
     def  dataReceived( self , data):
         self .transport.write(data)
 
def  main():
     factory  =  protocol.ServerFactory()
     factory.protocol  =  Echo
 
     reactor.listenTCP( 1234 ,factory)
     reactor.run()
 
if  __name__  = =  '__main__' :
     main()

　　

EchoClient

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from  twisted.internet  import  reactor, protocol
 
 
# a client protocol
 
class  EchoClient(protocol.Protocol):
     """Once connected, send a message, then print the result."""
 
     def  connectionMade( self ):
         self .transport.write( "hello alex!" )
 
     def  dataReceived( self , data):
         "As soon as any data is received, write it back."
         print  "Server said:" , data
         self .transport.loseConnection()
 
     def  connectionLost( self , reason):
         print  "connection lost"
 
class  EchoFactory(protocol.ClientFactory):
     protocol  =  EchoClient
 
     def  clientConnectionFailed( self , connector, reason):
         print  "Connection failed - goodbye!"
         reactor.stop()
 
     def  clientConnectionLost( self , connector, reason):
         print  "Connection lost - goodbye!"
         reactor.stop()
 
 
# this connects the protocol to a server running on port 8000
def  main():
     f  =  EchoFactory()
     reactor.connectTCP( "localhost" ,  1234 , f)
     reactor.run()
 
# this only runs if the module was *not* imported
if  __name__  = =  '__main__' :
     main()

运行服务器端脚本将启动一个TCP服务器，监听端口1234上的链接。服务器采用的是Echo协议，数据经TCP transport对象写出。运行客户端脚本将对服务器发起一个TCP链接，回显服务器端的回应而后终止链接并中止reactor事件循环。这里的Factory用来对链接的双方生成protocol对象实例。两端的通讯是异步的，connectTCP负责注册回调函数到reactor事件循环中，当socket上有数据可读时通知回调处理。

一个传送文件的例子 

server side 

client side 　　

　　

　　

Twisted深刻

http://krondo.com/an-introduction-to-asynchronous-programming-and-twisted/ 

http://blog.csdn.net/hanhuili/article/details/9389433 

　　

　　

SqlAlchemy ORM　　

SQLAlchemy是Python编程语言下的一款ORM框架，该框架创建在数据库API之上，使用关系对象映射进行数据库操做，简言之即是：将对象转换成SQL，而后使用数据API执行SQL并获取执行结果

Dialect用于和数据API进行交流，根据配置文件的不一样调用不一样的数据库API，从而实现对数据库的操做，如：

1
2
3
4
5
6
7
8
9
10
11
12
13
MySQL - Python
     mysql + mysqldb: / / <user>:<password>@<host>[:<port>] / <dbname>
  
pymysql
     mysql + pymysql: / / <username>:<password>@<host> / <dbname>[?<options>]
  
MySQL - Connector
     mysql + mysqlconnector: / / <user>:<password>@<host>[:<port>] / <dbname>
  
cx_Oracle
     oracle + cx_oracle: / / user: pass @host:port / dbname[?key = value&key = value...]
  
更多详见：http: / / docs.sqlalchemy.org / en / latest / dialects / index.html

　　

步骤一：

使用 Engine/ConnectionPooling/Dialect 进行数据库操做，Engine使用ConnectionPooling链接数据库，而后再经过Dialect执行SQL语句。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#!/usr/bin/env python
# -*- coding:utf-8 -*-
  
from  sqlalchemy  import  create_engine
  
  
engine  =  create_engine( "mysql+mysqldb://root:123@127.0.0.1:3306/s11" , max_overflow = 5 )
  
engine.execute(
     "INSERT INTO ts_test (a, b) VALUES ('2', 'v1')"
)
  
engine.execute(
      "INSERT INTO ts_test (a, b) VALUES (%s, %s)" ,
     (( 555 ,  "v1" ),( 666 ,  "v1" ),)
)
engine.execute(
     "INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)" ,
     id = 999 , name = "v1"
)
  
result  =  engine.execute( 'select * from ts_test' )
result.fetchall()

　　

步骤二：

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 进行数据库操做。Engine使用Schema Type建立一个特定的结构对象，以后经过SQL Expression Language将该对象转换成SQL语句，而后经过 ConnectionPooling 链接数据库，再而后经过 Dialect 执行SQL，并获取结果。

增删改查

 

一个简单的完整例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from  sqlalchemy  import  create_engine
from  sqlalchemy.ext.declarative  import  declarative_base
from  sqlalchemy  import  Column, Integer, String
from   sqlalchemy.orm  import  sessionmaker
 
Base  =  declarative_base()  #生成一个SqlORM 基类
 
 
engine  =  create_engine( "mysql+mysqldb://root@localhost:3306/test" ,echo = False )
 
 
class  Host(Base):
     __tablename__  =  'hosts'
     id  =  Column(Integer,primary_key = True ,autoincrement = True )
     hostname  =  Column(String( 64 ),unique = True ,nullable = False )
     ip_addr  =  Column(String( 128 ),unique = True ,nullable = False )
     port  =  Column(Integer,default = 22 )
 
Base.metadata.create_all(engine)  #建立全部表结构
 
if  __name__  = =  '__main__' :
     SessionCls  =  sessionmaker(bind = engine)  #建立与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
     session  =  SessionCls()
     #h1 = Host(hostname='localhost',ip_addr='127.0.0.1')
     #h2 = Host(hostname='ubuntu',ip_addr='192.168.2.243',port=20000)
     #h3 = Host(hostname='ubuntu2',ip_addr='192.168.2.244',port=20000)
     #session.add(h3)
     #session.add_all( [h1,h2])
     #h2.hostname = 'ubuntu_test' #只要没提交,此时修改也没问题
     #session.rollback()
     #session.commit() #提交
     res  =  session.query(Host). filter (Host.hostname.in_([ 'ubuntu2' , 'localhost' ])). all ()
     print (res)

　　

 

更多内容详见：

    http://www.jianshu.com/p/e6bba189fcbd

    http://docs.sqlalchemy.org/en/latest/core/expression_api.html

注：SQLAlchemy没法修改表结构，若是须要可使用SQLAlchemy开发者开源的另一个软件Alembic来完成。

步骤三：

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 全部组件对数据进行操做。根据类建立对象，对象转换成SQL，执行SQL。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
#!/usr/bin/env python
# -*- coding:utf-8 -*-
  
from  sqlalchemy.ext.declarative  import  declarative_base
from  sqlalchemy  import  Column, Integer, String
from  sqlalchemy.orm  import  sessionmaker
from  sqlalchemy  import  create_engine
  
engine  =  create_engine( "mysql+mysqldb://root:123@127.0.0.1:3306/s11" , max_overflow = 5 )
  
Base  =  declarative_base()
  
  
class  User(Base):
     __tablename__  =  'users'
     id  =  Column(Integer, primary_key = True )
     name  =  Column(String( 50 ))
  
# 寻找Base的全部子类，按照子类的结构在数据库中生成对应的数据表信息
# Base.metadata.create_all(engine)
  
Session  =  sessionmaker(bind = engine)
session  =  Session()
  
  
# ########## 增 ##########
# u = User(id=2, name='sb')
# session.add(u)
# session.add_all([
#     User(id=3, name='sb'),
#     User(id=4, name='sb')
# ])
# session.commit()
  
# ########## 删除 ##########
# session.query(User).filter(User.id > 2).delete()
# session.commit()
  
# ########## 修改 ##########
# session.query(User).filter(User.id > 2).update({'cluster_id' : 0})
# session.commit()
# ########## 查 ##########
# ret = session.query(User).filter_by(name='sb').first()
  
# ret = session.query(User).filter_by(name='sb').all()
# print ret
  
# ret = session.query(User).filter(User.name.in_(['sb','bb'])).all()
# print ret
  
# ret = session.query(User.name.label('name_label')).all()
# print ret,type(ret)
  
# ret = session.query(User).order_by(User.id).all()
# print ret
  
# ret = session.query(User).order_by(User.id)[1:3]
# print ret
# session.commit()

外键关联

A one to many relationship places a foreign key on the child table referencing the parent.relationship() is then specified on the parent, as referencing a collection of items represented by the child

from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
1
2
3
4
5
6
7
8
9
<br data - filtered = "filtered" > class  Parent(Base):
     __tablename__  =  'parent'
     id  =  Column(Integer, primary_key = True )
     children  =  relationship( "Child" )
 
class  Child(Base):
     __tablename__  =  'child'
     id  =  Column(Integer, primary_key = True )
     parent_id  =  Column(Integer, ForeignKey( 'parent.id' ))

To establish a bidirectional relationship in one-to-many, where the “reverse” side is a many to one, specify an additional relationship() and connect the two using therelationship.back_populates parameter:

1
2
3
4
5
6
7
8
9
10
class  Parent(Base):
     __tablename__  =  'parent'
     id  =  Column(Integer, primary_key = True )
     children  =  relationship( "Child" , back_populates = "parent" )
 
class  Child(Base):
     __tablename__  =  'child'
     id  =  Column(Integer, primary_key = True )
     parent_id  =  Column(Integer, ForeignKey( 'parent.id' ))
     parent  =  relationship( "Parent" , back_populates = "children" )

Child will get a parent attribute with many-to-one semantics.

Alternatively, the backref option may be used on a single relationship() instead of usingback_populates:

1
2
3
4
class  Parent(Base):
     __tablename__  =  'parent'
     id  =  Column(Integer, primary_key = True )
     children  =  relationship( "Child" , backref = "parent" )

　　

　　

附，原生sql join查询

几个Join的区别 http://stackoverflow.com/questions/38549/difference-between-inner-and-outer-joins 

  • INNER JOIN: Returns all rows when there is at least one match in BOTH tables
  • LEFT JOIN: Return all rows from the left table, and the matched rows from the right table
  • RIGHT JOIN: Return all rows from the right table, and the matched rows from the left table
1
select  host.id,hostname,ip_addr,port,host_group. name  from  host  right  join  host_group  on  host.id = host_group.host_id

in SQLAchemy

1
session.query(Host). join (Host.host_groups).filter(HostGroup. name == 't1' ).group_by( "Host" ). all ()

　　

group by 查询

1
select  name , count (host.id)  as  NumberOfHosts  from  host  right  join  host_group  on  host.id= host_group.host_id  group  by  name ;

in SQLAchemy

1
2
3
4
5
6
from  sqlalchemy import func
session.query(HostGroup, func. count (HostGroup. name  )).group_by(HostGroup. name ). all ()
 
#another example
session.query(func. count ( User . name ),  User . name ).group_by( User . name ). all ()  SELECT  count (users. name )  AS  count_1, users. name  AS  users_name
FROM  users  GROUP  BY  users. name

　　

 　　

　　

　　

 

更多ORM内容猛点这里

 

 

 

本节做业一

题目:IO多路复用版FTP

需求:

  1. 实现文件上传及下载功能
  2. 支持多链接并发传文件
  3. 使用select or selectors

 

本节做业二

题目：rpc命令端

需求：

  1. 能够异步的执行多个命令
  2. 对多台机器

>>:run "df -h" --hosts 192.168.3.55 10.4.3.4 task id: 45334>>: check_task 45334 >>:

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程