Python 10：线程、进程、协程、异步io

Python 10：线程、进程、协程、异步io

---

1、概念

2、线程

3、进程

4、协程

5、事件件驱动模型

6、异步io

7、作业

---

 一、概念

1、进程：

程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。

（程序以一个整体的形式暴露给操作系统。里面包含对各种资源的调用，内存的管理，网络接口的调用等，即对各种资源管理的集合）

程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。

启动一个进程会自动启动一个线程，进程里的第一个线程就是主线程，主线程可以创建子线程

2、线程：

线程是操作系统能够进行运算调度的最小单位，是一串指令的集合。

它被包含在进程之中，是进程中的实际运作单位。

一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务

进程要操作cpu，至少创建一个线程

所有在同一个进程里的线程共享同一块内存空间

3、区别：

线程共享内存空间，进程内存是独立的

多个线程可以直接访问同一个进程的数据，多个子进程之间数据是相互独立的

同一个进程的线程之间可以直接交流，两个进程想通信必须通过一个中间代理来实现

创建新的线程很简单，创建新的进程需要对其父进程进行一次克隆

一个线程可以控制和操作同一进程里的其他线程，但是进程只能操作子进程

对于主线程的修改有可能会影响到其他线程的运行，但是对一个父进程的修改不会影响其他子进程 

二、线程

1、调用

import threading
import time
def run(n):
    print("task",n)
    time.sleep(2)
t1 = threading.Thread(target=run,args=("t1",))
t2 = threading.Thread(target=run,args=("t2",))
t1.start()   #并行
t2.start()
# run("t1")  #未采用多线程：串行
# run("t2")

import threading
class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()
        self.n = n
    def run(self):
        print("running task",self.n)
t1 = MyThread("t1")
t2 = MyThread("t2")
t1.start()
t2.start()

2、 Join 和 Daemon 

import threading
import time
def run(n):
    print("task",n,threading.current_thread())   #<Thread(Thread-1, started 1516)>
    time.sleep(2)
start_time = time.time()
t_objs = []
for i in range(10):
    t = threading.Thread(target=run,args=("t-%s"%i,))
    t.start()
    t_objs.append(t)
for t in t_objs:
    t.join()     #在主线程里等待子线程结果
print("who:",threading.current_thread(),threading.active_count())  #打印当前线程 <_MainThread(MainThread, started 6568)> 和当前活动线程数
print("cost:",time.time() - start_time) #为加join主线程与子线程并行，不会等子线程运行结束

import threading
import time
def run(n):
    print("task",n,threading.current_thread())   #<Thread(Thread-1, started 1516)>
    time.sleep(2)
start_time = time.time()
t_objs = []
for i in range(10):
    t = threading.Thread(target=run,args=("t-%s"%i,))
    t.setDaemon(True)  #把当前线程设置为守护线程（主进程结束时不会等待守护线程执行完毕）
    t.start()  #一定要在执行之前设置
    t_objs.append(t)
# for t in t_objs:
#     t.join()     #在主线程里等待子线程结果
print("who:",threading.current_thread(),threading.active_count())  #打印当前线程 <_MainThread(MainThread, started 6568)> 和当前活动线程数
print("cost:",time.time() - start_time) #为加join主线程与子线程并行，不会等子线程运行结束

#程序正常退出会等待所有进程执行完毕（守护进程除外）

3、GIL

全局解释器锁 ：无论你启多少个线程，你有多少个cpu, Python在执行的时候在同一时刻只允许一个线程运行。

4、线程锁(互斥锁Mutex)

一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，此时，如果2个线程同时要修改同一份数据，有可能线程1还没修改完成但是线程2已经修改完成，为了避免线程自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。 

import threading
def addNum():
    lock.accquire()                 #加锁：只允许一个线程修改数据，在这里把线程变成了串行
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    num  +=1 #对此公共变量进行+1操作
    lock.release()                 #释放锁
lock = threading.Lock()         #申请锁
num = 0  #设定一个全局变量
thread_list = []
for i in range(10):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
for t in thread_list: #等待所有线程执行完毕
    t.join()
print('final num:', num )

5、递归锁

就是在一个大锁中还要再包含子锁，即有多个锁同时存在，当释放锁的时候系统自己混乱对不上号

递归锁的存在让程序能分辨出每个锁所对应的钥匙

import threading,time

def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num +=1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global  num2
    num2+=1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res,res2)


if __name__ == '__main__':

    num,num2 = 0,0
    lock = threading.RLock()   #递归锁（）
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()

while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num,num2)

6、Semaphore(信号量)

互斥锁 ：同时只允许一个线程更改数据，而Semaphore是同时只允许一定数量的线程更改数据

import threading,time

def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s
" %n)
    semaphore.release()

semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
for i in range(20):
    t = threading.Thread(target=run,args=(i,))
    t.start()
while threading.active_count() != 1:
    pass
else:
    print('----all threads done---')

7、event（线程交互）

通过Event来实现两个或多个线程间的交互

import time
import threading

event = threading.Event()
def light():
    count = 0
    event.set()
    while True:
        if count > 5 and count <= 10: #变成红灯
            event.clear()  #清空标志位
            print("33[41;1mred light is on ...33[0m")
        elif count > 10:
            event.set()  #变绿灯
            count = 0
        else:
            print("33[42;1mgreen light is on ...33[0m")
        time.sleep(1)
        count += 1

def car(name):
    while True:
        if event.is_set():  #判断是否设置了标志位
            print("[%s] running ..."%name)
            time.sleep(1)
        else:
            print("[%s] see red light ,waiting..."%name)
            event.wait()
            print("[%s] green light is on start going ..."%name)

light = threading.Thread(target=light,)
light.start()
car1 = threading.Thread(target=car,args=("car_1",))
car1.start()

8、queue队列 

import queue
q = queue.Queue()   #先入先出队列
# q = queue.Queue(maxsize=2) #队列接受最大容量
q.put("a")   #往队列里放入数据
q.put("b")
q.put("c")
print(q.qsize() )    #队列大小
q.get()    #从队列里取数据
q.get()

qq = queue.LifoQueue()  #后进先出对列
qq.put("a")   #往队列里放入数据
qq.put("b")
qq.put("c")
print(qq.get())  #c

qqq = queue.PriorityQueue()  #存储数据时可设置优先级的队列(vip)
qqq.put((9,"a"))  #放入元组,队列将按照元组第一个值进行排序
qqq.put((5,"b"))
qqq.put((7,"c"))
qqq.put((3,"d"))
print(qqq.get())   #(3, 'd')
print(qqq.get())   #(5, 'b')

9、生产者消费者模型

import threading,time
import queue
def producer():
    count = 1
    while True:
        q.put("骨头 %s" % count )
        print("生产了骨头%s"%count)
        count += 1
        time.sleep(2)
def consumer(n):
    # while q.qsize() >0:
    while True:
        print("%s 取到" %n, q.get())
        time.sleep(1)

q = queue.Queue(maxsize=10)
p = threading.Thread(target=producer,)
p.start()
c1 = threading.Thread(target=consumer,args=("aa",))
c1.start()
c2 = threading.Thread(target=consumer,args=("bb",))
c2.start()

 三、进程

在python中由于GIL的存在，无论你启多少个线程，你有多少个cpu, Python在执行的时候在同一时刻只允许一个线程运行

所以python的多线程不适合cpu密集操作型的任务，适合io密集型的任务（io操作不占用cpu、计算占用cpu）

注意：如果在windows中启动多进程必须采用：if __name__ =="__main__"（区分是自己主动执行还是被其他程序调用）

1、调用 

import multiprocessing,threading
import time

def run(name):
    time.sleep(1)
    print("hello",name)
    t = threading.Thread(target=thread_run())
    t.start()

def thread_run():
    print(threading.get_ident())

if __name__ == '__main__':
    for i in range(10):
        p = multiprocessing.Process(target=run,args=("aa",))
        p.start()

2、父进程、子进程：每一个进程都是由父进程启动的

from multiprocessing import Process
import os
def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())
    print("

")
def f(name):
    info('33[31;1mfunction f33[0m')
    print('hello', name)
if __name__ == '__main__':
    info('33[32;1mmain process line33[0m')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

3、进程间通信：

3.1进程间数据的传递

from multiprocessing import Process, Queue
def f(q):
    q.put([42, None, 'hello'])
if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"
    p.join()

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello from child'])
    print(conn.recv())
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()   #管道一生成就会产生两个返回对象
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    parent_conn.send("hello from parent")
    p.join()

3.2进程间数据的共享

from multiprocessing import Process, Manager
import os
def f(d, l):
    d[os.getpid()] = os.getpid()
    l.append(os.getpid())
    print(l)
 
if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()  #生成一个字典可在多个进程间传递
        l = manager.list(range(5))  #生成一个列表可在多个进程间传递
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:  #等待结果
            res.join()
        print(d)
        print(l)

4、进程锁：因为屏幕是共享的，多个进程可能同时使用屏幕输出，防止输出错乱而加锁

from multiprocessing import Process, Lock
 
def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release() 
if __name__ == '__main__':
    lock = Lock() 
    for num in range(10):
        Process(target=f, args=(lock, num)).start()

5、进程池

from  multiprocessing import Process,Pool
import time
import os
def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i+100

def Bar(arg):
    print('-->exec done:',arg)
if __name__ == "__main__":
    pool = Pool(5)
    for i in range(10):
        pool.apply_async(func=Foo, args=(i,),callback=Bar)   #apply_async:并行  callback=回调(主进程回调，在主进程中执行)
        #pool.apply(func=Foo, args=(i,))          #apply:串行
    print('end')
    pool.close() #必须先close在join
    pool.join()#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

 四、协程

协程，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程（CUP根本不知道他的存在）

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。

优点：

无需线程上下文切换的开销

无需原子操作锁定及同步的开销

方便切换控制流，简化编程模型

高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

 缺点：

无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。

进程阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

协程一个标准定义：

1、必须在只有一个单线程里实现并发

2、修改共享数据不需加锁

3、用户程序里自己保存多个控制流的上下文栈

4、一个协程遇到IO操作自动切换到其它协程

greenlet

#greenlet 自动版的yield
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()

greenlet虽然满足了协程定义前3个条件但是没有遇到IO操作自动切换功能。

gevent

import gevent
 
def func1():
    print('run in func1')
    gevent.sleep(2)
    print('run in func1_1')
def func2():
    print('run in func2')
    gevent.sleep(1)
    print('run in func2_2')
def func3():
    print('run in func3')
    gevent.sleep(1.5)
    print('run in func3_3')
gevent.joinall([
    gevent.spawn(func1),
    gevent.spawn(func2),
    gevent.spawn(func3),
])

python可以通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。

举个栗子

from gevent import monkey #让gevent知道url在进行io操作
import gevent
from  urllib.request import urlopen

monkey.patch_all()  #把当前程序的所有io操作给我单独的做上标记
def f(url):
    print('GET: %s' % url)
    resp = urlopen(url)
    data = resp.read()
    f = open('url.html','wb')
    f.write(data)
    f.close()
    print('%d bytes received from %s.' % (len(data), url))
 
gevent.joinall([
        gevent.spawn(f, 'https://www.baidu.com/'),
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])

五、事件驱动模型

通常，我们写服务器处理模型的程序时，有以下几种模型：

1、每收到一个请求，创建一个新的进程，来处理该请求：创建新的进程的开销比较大，所以，会导致服务器性能比较差,但实现比较简单。

2、每收到一个请求，创建一个新的线程，来处理该请求：要涉及到线程的同步，有可能会面临死锁等问题

3、每收到一个请求，放入一个事件列表，让主进程通过非阻塞I/O方式来处理请求：在写应用程序代码时，逻辑比前面两种都复杂

综合考虑各方面因素，一般普遍认为第3种方式是大多数网络服务器采用的方式

1、事件驱动模型：

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

2、编程范式：

事件驱动编程：这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。

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

单线程编程：

在单线程模式中，任务按照顺序执行。如果某个任务因为I/O而阻塞，其他所有的任务都必须等待，直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系，但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。

多线程编程：

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

当我们面对如下的环境时，事件驱动模型通常是一个好的选择：

1、程序中有许多任务，而且io调用较多

2、任务之间高度独立（因此它们不需要互相通信，或者等待彼此）

3、在等待事件到来时，某些任务会阻塞。

网络应用程序通常都有上述这些特点，这使得它们能够很好的契合事件驱动编程模型。

 六、异步IO

1、io模式

对于一次IO访问（以read举例），数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。当一个read操作发生时，它会经历两个阶段：

1. 等待数据准备 (Waiting for the data to be ready)

2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正式因为这两个阶段，linux系统产生了下面五种网络模式的方案。

阻塞 I/O（blocking IO）

非阻塞 I/O（nonblocking IO）

I/O 多路复用（ IO multiplexing）

信号驱动 I/O（ signal driven IO）

异步 I/O（asynchronous IO）

2、I/O 多路复用之select、poll、epoll

select，poll，epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。

import select
import socket
import queue
server = socket.socket()
server.bind(("localhost",9000))
server.listen(5)
server.setblocking(False)
msg_dic = {}
inputs = [server,]
outputs =[]  #r
while True:
    readable,writeable,exceptional = select.select(inputs,outputs,inputs)
    print(readable,writeable,exceptional)
    for r in readable:
        if r is server:  #代表来了一个新连接
            conn,addr = server.accept()
            print("来了个新连接",addr)
            conn.setblocking(0)
            inputs.append(conn) #因为这个新建立的连接还没发数据过来，现在就接收的话。程序就会报错了，
            #所以要想实现这个客户端发数据来时server端能知道，就需要让select在检测这个conn
            msg_dic[conn] = queue.Queue() #初始化一个队列，后面存要返回给客户端的数据
        else:
            try:
                data = r.recv(1024)
                if data:
                    print("收到数据",data)
                    msg_dic[r].put(data)
                    outputs.append(r)  #放入返回的连接队列里
                else:#如果收不到data代表代表客户端断开了
                    print("客户端断开了",r)
                    if r in outputs:
                        outputs.remove(r) #清理已断开的连接
                    inputs.remove(r) #清理已断开的连接
                    del msg_dic[r] ##清理已断开的连接
            except ConnectionResetError as e:
                print("客户端断开了",r)
                if r in outputs:
                    outputs.remove(r) #清理已断开的连接
                inputs.remove(r) #清理已断开的连接
                del msg_dic[r] ##清理已断开的连接
    for w in writeable:  #要返回给客户端的连接列表
        data_to_client = msg_dic[w].get()
        w.send(data_to_client)  #给客户返回原数据
        outputs.remove(w)  #确保下次循环的时候，writeable不返回这个已经处理完的连接
    for e in exceptional: #连接断开
        if e in outputs:
            outputs.remove(e)
        inputs.remove(e)
        del msg_dic[e]

import socket

HOST = 'localhost'
PORT = 9000
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("received",data)

s.close()

3、selectors模块 

import selectors
import socket
sel = selectors.DefaultSelector()
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)  #新连接注册read回调函数
def read(conn, mask):
    data = conn.recv(1024)  # Should be ready
    if data:
        print('echoing', repr(data), 'to', conn)
        conn.send(data)  # Hope it won't block
    else:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()
sock = socket.socket()
sock.bind(('localhost', 8000))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)   #只要来新连接就调用accept
while True:
    events = sel.select()  #默认阻塞，有活动链接就返回活动的连接列表
    for key, mask in events:
        callback = key.data   #调用accept
        callback(key.fileobj, mask)  #key.fileobj = 文件句柄

七、作业

题目:IO多路复用版FTP

需求:

1、实现文件上传及下载功能

2、支持多连接并发传文件

3、使用select or selectors

selector版

import os,sys,json,time
import selectors
import socket
import errno

BASE_DIR  = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)
sel = selectors.DefaultSelector()
class selectors_ftp(object):
    def __init__(self,ip,port):
        self.sock = socket.socket()
        self.sock.bind((ip,port))
        self.sock.listen(500)
        self.sock.setblocking(False)
        self.start(self.sock)

    def start(self,sock):
        sel.register(sock, selectors.EVENT_READ, self.accept)
        while True:
            events = sel.select()
            for key, mask in events:
                callback = key.data
                callback(key.fileobj,mask)
    def accept(self,sock,mask):
        self.conn, self.addr = sock.accept()
        print( "有新连接：",self.addr)
        self.conn.setblocking(False)
        sel.register(self.conn, selectors.EVENT_READ, self.read)

    def read(self,conn,mask):
        try:
            print("start..")
            time.sleep(2)
            self.data = conn.recv(1024)
            if self.data:
                self.data_receive = json.loads(self.data.decode())
                self.action = self.data_receive['action']
                self.file_name = self.data_receive['file_name']
                self.file_size = self.data_receive['size']
                print('有新的数据：',self.data_receive)
                if hasattr(self,self.action):
                    func = getattr(self,self.action)
                    func(self.conn,mask)
            else:
                print('连接断开！')
                sel.unregister(conn)
                conn.close()
        except ConnectionResetError as e:
            print('连接断开！',e)
            sel.unregister(conn)
            conn.close()

    def put(self,conn,mask):
        self.conn.send("收到文件，正在上传...".encode())
        file_object = open((self.file_name+"_new"), 'wb')
        received_size = 0
        while received_size < self.file_size:
            try:
                if self.file_size - received_size > 1024:
                    size = 1024
                elif self.file_size < 1024:
                    size = self.file_size
                else:
                    size = self.file_size - received_size
                recv_data = conn.recv(size)
                received_size += len(recv_data)
                file_object.write(recv_data)
            except BlockingIOError as e:
                if e.errno != errno.EAGAIN:  # errno.EAGAIN 缓冲区满
                    raise
            else:
              time.sleep(0.1)
        else:
            file_object.close()
        print("上传完毕")
        conn.send("文件上传完毕".encode())

    def get(self,conn,mask):
        while True:
            if os.path.isfile(self.file_name):
                try:
                    file_size = os.path.getsize(self.file_name)
                    self.conn.send(str(file_size).encode())
                    client_file_size = 0
                    with open(self.file_name, "rb") as file_obj:
                        for line in file_obj:
                            client_file_size += len(line)  # 记录已经传送的文件大小
                            self.conn.sendall(line)
                    file_obj.close()
                    if client_file_size >= int(file_size):  # 文件传送完毕
                        break
                except BlockingIOError as e:
                    if e.errno != errno.EAGAIN:  # errno.EAGAIN 缓冲区满 等待下次发送
                        raise
                else:
                    time.sleep(0.1)  # 等待进行下一次读取
            else:
                conn.send(b'error')
                break

s = selectors_ftp("localhost",8000)

import os,sys,json,time
import socket

BASE_DIR  = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)

class selectors_client(object):
    def __init__(self,ip,port):
        self.client = socket.socket()
        self.client.connect((ip, port))
        self.start()
    def start(self):
        print('login in')
        while True:
            try:
                self.sending_msg_list = []
                self.sending_msg = input('>>>:')
                self.sending_msg_list = self.sending_msg.split()
                self.action = self.sending_msg_list[0]
                if len(self.sending_msg_list) == 0:
                    continue
                elif len(self.sending_msg_list) == 1:
                    if self.sending_msg_list[0] == "exit":
                        print('logout')
                        break
                    else:
                        print('command not found' )
                else:
                    try:
                        self.file_name = self.sending_msg_list[1]
                    except IndexError:
                        pass
                    if hasattr(self,self.action):
                        func = getattr(self,self.action)
                        func()
                    else:
                        print( 'command not found')
            except ConnectionResetError and ConnectionRefusedError and OSError and IndexError as e:
                print( e,'Restart client')
                selectors_client().start()
    def put(self):
         if  os.path.isfile(self.file_name):
             self.file_size = os.path.getsize(self.file_name)
             data_header = {
                 "action": "put",
                 "file_name": self.file_name,
                 "size": self.file_size}
             self.client.send(json.dumps(data_header).encode())
             print(self.client.recv(1024).decode())
             with open(self.file_name, 'rb') as file_object:
                 for line in file_object:
                     self.client.send(line)
                 file_object.close()
             print(self.client.recv(1024).decode())
         else:
             print('No such file')
    def get(self):
        data_header = {
            "action": "get",
            "file_name": self.file_name,
            "size": 0}
        self.client.send(json.dumps(data_header).encode())
        self.data = self.client.recv(1024)
        if self.data.decode() == 'error':
            print('No such file')
        else:
            print("文件大小:", self.data.decode())
            file_object = open((self.file_name + '_new' ), 'wb')
            received_size = 0
            file_size = int(self.data.decode())
            while received_size < file_size:
                if file_size - received_size > 1024:
                    size = 1024
                elif file_size < 1024:
                    size = file_size
                else:
                    size = file_size - received_size
                recv_data = self.client.recv(size)
                received_size += len(recv_data)
                file_object.write(recv_data)
            else:
                file_object.flush()
                file_object.close()
                time.sleep(0.1)
                print("File get done File size is :", file_size)

c = selectors_client("localhost",8000)