Python的进程和线程

线程和进程

---

概念

什么是进程：进程就是一个过程，一个任务。进程和线程，都是一个任务而已。。。

什么是并行和并发：CPU同一时刻只能执行一个任务。 无论是并行还是并发，都是让用户看来是在同时运行，多进程就是CPU快速的在不同的进程间来回切换。

1. 并行：并行就是多个CPU同时运行多个任务（每个CPU运行一个任务），或者单个CPU包含多个核心。 有多少个CPU或多少个核心，就可以同时执行多少个任务。
2. 并发：伪并行，是指CPU在多道程序间来回切换，实现的多个程序同时运行的效果。

同步与异步的概念：

1. 同步：就是当一个进程在执行一个请求时，如果需要等待一段时间得到返回信息时，那么该进程一直等待，直到收到返回消息后才继续执行。
2. 异步：就是当一个进程在执行一个请求时，不管其他进程的状态，不等待返回结果继续向下执行，当有消息返回时系统会发出通知，此时再进行处理。很大的提升了效率

进程的创建： 用户创建的进程，都是由系统来负责的。无论采用什么方式创建的进程，都是调用操作系统的接口来创建的，进程的切换都是由操作系统控制的。 无论是哪一种创建进程的方式，新的进程创建，都是由一个已经存在的进程执行了一个用于创建进程的系统调用得到的。

父进程和子进程：子进程创建后和父进程都有各自独立的地址空间，进程是通过多道技术使每个进程在物理层面上隔离的。任何一个进程在其地址空间的休息都不会影响到另外一个进程。 注意：子进程和父进程之间是可以有只读的共享的内存区域的。 进程与进程之间数据（资源）是隔离的，两个进程之间可以基于管道这种方式进行通信。在Unix当中，是含有进程层次的概念的，但是在windows当中，是没有进程层次的概念的，所有的进程都是地位相同的。 在Linux当中，每启动一个命令，都会启动一个进程。

什么是线程：一个进程里面至少有一个控制线程，进程的概念只是一种抽象的概念，真正在CPU上面调度的是进程 里面的线程,就像是铁路系统一样，铁路系统内至少有一条铁路线（线程），铁路线是跑火车的（实际工作的），火车共享着铁路系统提供的资源（使用铁路的设施）。也就是说，进程是一个抽象的概念，更像是一个容器，里面包含很多干活的线程，这些线程共享着同一个进程里包含的所有资源，而线程是一个调度单位，不包含资源。

什么时候需要开启多个线程：一个进程里面的多个线程共享这个进程里面的资源，因此如果多个任务共享同一块资源的时候，需要开启多个线程。多线程指的是，在一个进程中开启多个线程。 简单的说：如果多个任务共用同一个资源空间，那么必须在一个进程内开启多个线程。

多线程和多进程的应用方向：

1. 对于计算密集型应用，应该使用多进程;
2. 对于IO密集型应用，应该使用多线程。 线程的创建比进程的创建开销小的多。

Python中线程的特点： 在其他语言当中，一个进程里面开启多个线程，每个线程都可以给一个cpu去使用，但是在python当中，在同一时刻，一个进程当中只能有一个线程处于运行状态。 eg:在其他语言当中，比如我现在开启了一个进程，这个进程当中含有几个线程，如果我现在有多个cpu，每一个线程是可以对应相应的CPU的。 但是在python当中，如果我们现在开启了一个进程，这个进程里面对应多个线程，同一时刻只有一个线程可以处于运行状态。对于其他语言而言，在多CPU系统中，为了最大限度的利用多核，可以开启多个线程。 但是Python中的多线程是利用不了多核优势的。

进程与线程的区别：

在同一个进程当中，多个线程彼此之间可以相互通信；但是进程与进程之间的通信必须基于IPC这种消息的通信机制（IPC机制包括队列和管道）。 在一个进程当中，改变主线程可能会影响其它线程的行为，但是改变父进程并不会影响其它子进程的行为，因为进程与进程之间是完全隔离的。 在python当中，在同一时刻同一进程当中只能同时有一个线程在运行，如果有一个线程使用了系统调用而阻塞，那么整个进程都会被挂起。

进程举例

开启子进程

#方法一：
from multiprocessing import Process
import os
import time


def task(msg):
    print('process:%s PID:%s PPID:%s ' % (msg, os.getpid(), os.getppid()))


if __name__ == '__main__':
    for i in range(3):
        p = Process(target=task, args=('进程%s' % i,))  #创建进程对象
        p.start()   #开启子进程

    print('主')

#方法二：
from multiprocessing import Process
import os


class MyProcess(Process):   #使用类的方法开启子进程，需要继承Process模块
    def __init__(self, msg):
        super().__init__()   #从用父类的__init__方法
        self.msg = msg

    def run(self):   #必须使用run这个名字，固定模式。
        print('process:%s PID:%s PPID:%s ' % (self.msg, os.getpid(), os.getppid()))


if __name__ == '__main__':
    for i in range(3):
        p = MyProcess('进程%s' % i)
        p.start()

守护进程

from multiprocessing import Process
import time
import os


def task(msg):
    print('this is daemon %s PID:%s start' % (msg, os.getpid()))
    time.sleep(2)   #模拟运行时间
    print('daemon %s PID:%s is done.' % (msg, os.getpid()))


if __name__ == '__main__':
    p = Process(target=task, args=('守护',))
    p.daemon = True   #设置为守护进程
    p.start()
    time.sleep(1)  模拟主进程运行时间，当1秒过后主进程结束守护进程一起死掉。
    print('主')
#当主进程的运行时间超过守护进程的时间，就可以看到子进程运行结束的信息输出了。

互斥锁

from multiprocessing import Process, Lock
import os
import time


def task(name, mutex):
    mutex.acquire() #开启互斥锁
    print('the process %s pid:%s' % (name, os.getpid()))
    time.sleep(1)
    mutex.release() #释放互斥锁


if __name__ == '__main__':
    mutex = Lock() #创建互斥锁
    for i in range(3):
        p = Process(target=task, args=('子进程%s' % i, mutex,)) #进程互斥锁需要作为参数带入
        p.start()
    print('主')
#进程谁先抢到互斥锁谁就先执行，当执行完毕释放互斥锁后其他进程才能再得到互斥锁来执行。

互斥锁和JOIN的区别：

1. 互斥锁：对需要共享数据的局部代码进行串行执行
2. join：对程序整体进行串行

进程Queue

进程Queue，即队列，可以存放任意类型的数据。是用来在多个进程间通讯使用的。队列不宜存放大数据，适合存放精简的小数据。队列占用的是内存空间，应该适当设置队列大小。

通过生产者消费者模型可以很容易的理解队列的作用（解耦合），但使用MultiProcess的队列，会使程序都在一个主机上运行，这就形成了集中式，影响稳定性，性能局限（不便于拓展）

#生产者消费者模型 
from multiprocessing import Process, Queue
import time


def producer(q):    #Q为队列对象
    for i in range(5):
        res = '产品%s' % i   #生产的的内容 
        time.sleep(1)
        print('生产了%s' % res)
        q.put(res)     #将生产的内容放入队列


def consumer(q):  
    while True: #使用while是因为不位置生产的个数。
        res = q.get()   #获得队列中的内容
        if res is None:  #判断队列是否为空，空就结束当前进程。
            break
        time.sleep(1.5)
        print("获得了%s" % res)


if __name__ == '__main__':
    q = Queue()  #创建队列
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=producer, args=(q,))
    p3 = Process(target=producer, args=(q,))
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))
    #以上是3个生产者进程，和2个消费者进程
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()
    #同时开启5个进程
    p1.join()
    p2.join()
    p3.join()
    #等待3个生产进程全部结束
    q.put(None)
    q.put(None)
    #有几个消费者就要有几个PUT(NONE)。给消费者提供结束信号。

    print('主进程')

JoinableQueue 用法

#生产者消费者模型
from multiprocessing import Process, JoinableQueue
import time


def producer(q):
    for i in range(5):
        res = '产品%s' % i
        time.sleep(1)
        print('生产了%s' % res)
        q.put(res)
    q.join()            #要在生产者中加入JOIN()，等待消费者发出信号。

def consumer(q):
    while True:
        res = q.get()
        if res is None:
            break
        time.sleep(1.5)
        print("获得了%s" % res)
        q.task_done()   #发出信号给join()告知已经取完队列中所有信息

if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=producer, args=(q,))
    p3 = Process(target=producer, args=(q,))
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))

    c1.daemon = True        #守护进程
    c2.daemon = True

    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()  #当所有生产者运行完毕后，主进程结束，同时消费者作为守护进程也结束。

    print('主进程')

开启线程

开启线程与开启进程是一样的方法。

#方法一：
from threading import Thread
import os
import time


def task(msg):
    print('Thread:%s PID:%s PPID:%s ' % (msg, os.getpid(), os.getppid()))


if __name__ == '__main__':
    for i in range(3):
        p = Thread(target=task, args=('线程%s' % i,))  #创建线程对象
        p.start()   #开启线程

    print('主')

#方法二：
from threading import Thread
import os


class MyThread(Thread):   #使用类的方法开启线程，需要继承Thread模块
    def __init__(self, msg):
        super().__init__()   #从用父类的__init__方法
        self.msg = msg

    def run(self):   #必须使用run这个名字，固定模式。
        print('Thread:%s PID:%s PPID:%s ' % (self.msg, os.getpid(), os.getppid()))


if __name__ == '__main__':
    for i in range(3):
        p = MyThread('线程%s' % i)
        p.start()

线程对象中的方法：

currentThread() 返回当前线程的变量

currentThread().getName() 获取当前线程的名字

setName() 设置线程名

enumerate() 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。

active_count() 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

守护线程

from threading import Thread
import time


def t1():
    print('线程开启111')
    time.sleep(1)
    print('线程1111结束')


def t2():
    print('线程开启2222')
    time.sleep(2)
    print('线程2222结束')


if __name__ == '__main__':
    t1 = Thread(target=t1)
    t2 = Thread(target=t2)
    t1.daemon = True
    t1.start()
    t2.start()
    print('主线程')

和守护进程不一样，由于T2还没有运行结束，主线程等待所有线程结束后才能结束，这时T1线程由于比T2线程运行时间短，所以T1可以正常执行结束。（这与守护进程是不一样的。归根结底就是，进程是独立的，线程是共享同一进程的，所以守护进程和守护线程的不同之处就在这里）

线程互斥锁

from threading import Thread, Lock
import time

n =100

def task():
    global n
    mutex.acquire()
    time.sleep(0.1)  #这里可以不用睡0.1秒。因为线程是秒开的不需要开系统资源，但在CPYthon解释器
                     #以外的其他解释下执行，就需要加入等待时间了。因为CPYthon的GIL垃圾回收特性
                     #使得，同一个进程下的多个线程在同一时间只能运行一个，无法发挥计算机的多核优
                     #势，但在其他python解释下则不然。
    n -= 1
    mutex.release()


if __name__ == '__main__':
    mutex = Lock()   #定义互斥锁
    ls = []    

    for i in range(100):
        t1 = Thread(target=task, name='%s' % i) #name是设置了线程名
        t1.start()
        ls.append(t1)
        print(t1)

    for i in ls:  #利用循环和列表，对所有开启的线程进行JOIN的等待操作。
        i.join()

    print('主', n)

线程的互斥锁和进程的互斥锁并没有什么区别，主要的区别就在于，开线程是在同一个进程下开启的，线程会共用进程中的资源（进程是资源单位，线程是运行单位），而进程是系统通过调用接口来创建的，每个进程都会占用一定的系统资源，并且每个进程都是独立的。

GIL锁

GIL是CPython的特性，GIL是解释器层面的一把锁，保证了在同一时间只能有一个活跃的线程来运行PYTHON的字节码（python解释器代码），因为PYthon的内存垃圾回收机制不是安全的，要保证线程的数据安全，就需要用到这个GIL锁，并且PYthon的一些其他功能都会依赖于这把锁。

GIL锁是解释器层面的锁，针对的是垃圾回收功能，它不能保证数据的安全，所以要在GIL的基础上再加上LOCK（互斥锁）确保PYTHON在同一时间只执行一个任务的代码，来保证数据安全。

注：在CPYthon解释器中，同一个进程下开启的多个线程，同一时间只能有一个线程执行，无法利用多核优势！

GIL与多线程

1. 对于计算来说，CPU越多越好，但是对于I/O来说，再多的CPU也没用。
2. 如果是单核CPU，程序用线程设计，因为对于单核CPU来说，多进程没有意义。
3. 如果是多核CPU，面对计算密集型来说要用到多进程，这样计算会很快。（金融类）
4. 如果是多核CPU，面对I/O密集型来说，要用到多线程，因为多进程与多线程就、效率几乎一样。（网络类）

os.cpu_count()方法 查看计算机CPU核心数

死锁与递归锁

#死锁模拟
from threading import Thread, Lock
import time

mutexA = Lock()
mutexB = Lock()


class MyThread(Thread):  #通过类的方式开启线程

    def run(self):  #run方法名为固定格式
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到A锁' % self.name)
        mutexB.acquire()
        print('%s 拿到B锁' % self.name)

        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('%s 拿到B锁' % self.name)
        time.sleep(0.1)  #设置0.1秒的等待以便其他线程都启动起来。
        mutexA.acquire()
        print('%s 拿到A锁' % self.name)

        mutexA.release()
        mutexB.release()


if __name__ == '__main__':
    for i in range(10):
        t = MyThread()
        t.start()

这段代码，由于线程1在F2中拿到B锁，但无法拿到A锁，因为线程1在F2中拿到B锁后，睡了0.1秒，这时线程2在F1中拿到了A锁，当线程1,0.1秒时间过后，再去拿A锁，这时A锁已经被线程2占用，而线程2在F1中想拿B锁，可B锁被线程1占用，这时就形成了死锁。

#递归锁
from threading import Thread, RLock
import time

mutexA = mutexB = RLock()    #A和B都等于递归锁


class MyThread(Thread):   #继承Thread类

    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到A锁' % self.name)
        mutexB.acquire()
        print('%s 拿到B锁' % self.name)

        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('%s 拿到B锁' % self.name)
        time.sleep(0.1)
        mutexA.acquire()
        print('%s 拿到A锁' % self.name)

        mutexA.release()
        mutexB.release()


if __name__ == '__main__':      #只要递归锁不为0，其他线程就不能执行。
    for i in range(10):
        t = MyThread()
        t.start()

递归锁是将多把锁连接在一起，就像钥匙串一样，当一个线程拿到钥匙串中的任意一把锁或多把锁时，其他的线程都不能拿到该钥匙串中的任意锁，直到该线程将所有的锁释放掉后才可以被其他的线程拿到递归锁。（没得到一把锁就在队列中加一，释放一把锁就字啊队列中减一，直到全部释放为止）

信号量

from threading import Thread, Semaphore, current_thread
import time

sm = Semaphore(3)


def task():
    # sm.acquire()
    # print('%s in sm' % current_thread().name)
    # time.sleep(3)
    # sm.release()
    with sm:                                #另一种写法。。适用于互斥锁
        print('%s in sm' % current_thread().getName())
        time.sleep(3)


if __name__ == '__main__':
    for i in range(10):
        t = Thread(target=task)
        t.start()

信号量就是允许多少个线程同时运行，限制同时运行的线程数。

Event事件

from threading import Thread, Event
import time

event = Event()   #定义EVENT对象。


def student(name):
    print('%s 正在上课。' % name)
    event.wait()        #event.wait()等待SET被执行后执行下面语句
    event.wait(2)        #如果设置了时间，那么不等SET发出信号，只要到了时间后就会继续执行。
    print('%s 课间休息.' % name)


def teacher(name):
    print('%s 老师正在上课。。' % name)
    time.sleep(5)
    print('%s 老师下课..' % name)
    time.sleep(0.5)
    event.set()    #执行set后，wait解除等待，继续执行后面的代码。


if __name__ == '__main__':
    stu1 = Thread(target=student, args=('sly',))
    stu2 = Thread(target=student, args=('sj',))
    stu3 = Thread(target=student, args=('lby',))
    t1 = Thread(target=teacher, args=('good',))

    stu1.start()
    stu2.start()
    stu3.start()
    t1.start()

Event事件，就是在遇到WAIT进行等待，直到SET被执行后，执行WAIT后的代码。

event.isSet() 返回event的状态值。

event.wait() 如果event.isSet()==False 时将阻塞线程。

event.set() 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态，等待操作系统调度。

event.clear() 恢复event的状态值为False

定时器

#定时器验证码刷新认证:

from threading import Timer
import random
import string

class Test(object):
    def __init__(self):
        self.refresh()   #实例化时给出第一个验证码

    def refresh(self, time=5):  #刷新时间设置为默认参数
        self.res = self.get_code()
        print(self.res)
        self.t = Timer(time, self.refresh)  #使用定时器，等待time的时间后执行refresh方法。
        self.t.start()   #启动定时器



    def get_code(self):   #通过random类得到随机验证码
        code = ''.join(random.sample(string.digits + string.ascii_letters, 5))
        return code

    def check(self):
        while True:
            code = input('>>>').strip()
            if code == self.res:
                print('验证码正确!')
                self.t.cancel()    #cancel方法 是取消的意思。
                break

g = Test()
g.check()

定时器本身就是一个线程，所以与开线程一样来开定时器。

#小例子看定时器的线程ID和线程名
from threading import Timer, currentThread
import os


def refresh(time=2):
    print(msg(), currentThread().getName())
    t = Timer(time, refresh)
    t.start()

def msg():
    code = os.getpid()
    return code

if __name__ == '__main__':
    refresh()

通过这个小例子可以看到。PID都是同样的并且通过currentThread().getName()得到线程的名字。

线程Queue

import queue   #调用方法

q = queue.Queue(3)  #设置队列的容量
"""先进先出，队列"""
q.put('first')
q.put('second')
q.put('third')
# q.put('hh', block=True, timeout=2)  #默认block是为True阻塞状态, timeout是等待时间

print(q.get())
print(q.get())
print(q.get())
# q.get(block=True, timeout=2) #默认block是为True阻塞状态, timeout是等待时间
# q.get_nowait() #与block=False的效果是一样的。不等待。

q = queue.LifoQueue(3)  #设置堆栈的容量
"""后进先出，堆栈"""
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
#最先放入的数据最后被拿出来，最后放进的数据最先被拿出来。

q = queue.PriorityQueue(3)  #设置优先级的容量
"""优先级队列"""
q.put((10, 'first'))     #格式为元组的形式。前面的元素是优先级数字，越小优先级越高。
q.put((20, 'second'))
q.put((15, 'third'))

print(q.get())
print(q.get())
print(q.get())
#元组或列表形式存储，数字越小优先级越高，取值时正常。

池

#进程池和线程池（concurrent.futures模块)
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
import random
import os
from threading import currentThread

def task():
    print('%s PID:%s' % (currentThread().getName(), os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == '__main__':
    # pool = ProcessPoolExecutor(3)
    pool = ThreadPoolExecutor(3)  #进程池和线程池开启的方式一样

    for i in range(10):
        pool.submit(task)   #使用submit异步提交的方式提交到池中，由池进行开启

    pool.shutdown()   #使用shutdonw来关闭池的入口，池内线程执行完毕后结束。

    print('主线程', os.getpid())

池与信号量的区别

池的概念与信号量相似，都是允许同一时间有固定多个线程或进程同时执行。但池限定的数量不仅仅是同时执行的个数，而且是开启的线程数也是固定的。而信号量，虽然也是可以控制同时执行的个数，但信号量不会重复使用已开线程，而是开启新的线程。（长时间执行浪费系统资源）

异步调用与回调机制

from concurrent.futures import ThreadPoolExecutor
import time
import random


def pt(name):
    print('产出中。。。。。。')
    time.sleep(random.randint(1,3))
    res = random.randint(1, 10)
    result = {'name': name, 'res': res}
    print('%s 输出: %s 个' % (name, res))
    return result

def js(pt):
    pt = pt.result()   #必须使用result()来获得对象的返回值
    sum = int(pt['res']) * 27
    print(pt['name'], '获得:', sum)


if __name__ == '__main__':
    pool = ThreadPoolExecutor(5)

    pool.submit(pt, 'alex').add_done_callback(js)  #回调方法将pool.submit(pt, 'alex')有返回值时（是一个对象），
                                                   #将这个对象作为 参数 传给 add_done_callback()中的函数。
    pool.submit(pt, 'sly').add_done_callback(js)   #注意：回调函数执行后得到的是FUTURES对象。
    pool.submit(pt, 'cool').add_done_callback(js)
    pool.shutdown()

异步回调与线程并发的区别在于，SUBMIT执行后得到的结果（对象）立即作为adddonecallback()调用函数的参数被执行，然后得到返回结果。而线程并发是多线程一同执行一个函数后，再一同执行关联的函数。

协程

在单一线程下使用程序实现并发。用以欺骗CPU，从而为程序占用最大的CPU资源来提升程序的执行效率。

• 优点：1，写成的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级。2，单线程内就可以实现并发的效果，最大的限度地利用CPU。
• 缺点：1，协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程。2，协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程。

协程特点

1. 必须在只有一个单线程里实现并发。
2. 修改共享数据不需加锁
3. 用户程序里自己保存多个控制流的上下文栈
4. 附加：一个协程遇到I/O操作自动切换到其他协程（如何实现检测I/O，yield，greenlet都无法实现，就用到了gevent模块（select机制））

协程--greenlet模块

#通过pip3 install greenlet 来安装模块
from greenlet import greenlet


def eat(name):
    print('%s eat 1' % name)
    g2.switch('sly')
    print('%s eat 2' % name)
    g2.switch('sly')  #再次调用可以不用加入参数g2.switch()即可。

def play(name):
    print('%s play 1' % name)
    g1.switch('sly') #再次调用可以不用加入参数g1.switch()即可。
    print('%s play 2' % name)


g1 = greenlet(eat)
g2 = greenlet(play)

g1.switch('sly')  #首次调用的时候要加入参数，再次调用的时候可以不用加入参数。

该模块仅仅比yield好一点点，的确实现了程序间的来回切换，但无法检测阻塞。

协程--gevent模块

#使用pip3 install gevent 安装模块
from gevent import monkey  #gevent模块下的monkey类
monkey.patch_all()          
#调用monkey类下的patch_all()方法来检测代码中的阻塞，所以，一定要在最前面进行调用。
#from gevent import monkey;monkey.patch_all()  可以写成这样的格式。
import gevent
import time

def eat(name):
    print('%s eat 1' % name)
    time.sleep(3)
    print('%s eat 2' % name)


def play(name):
    print('%s play 1' % name)
    time.sleep(2)
    print('%s play 2' % name)


def read(name):
    print('%s read 1' % name)
    time.sleep(4)
    print('%s read 2' % name)


g1 = gevent.spawn(eat, 'sly')  #使用gevent.spawn创建协程对象来提交任务。
g2 = gevent.spawn(play, 'lby')
g3 = gevent.spawn(read, 'good')

# g1.join()
# g2.join()
# g3.join()

gevent.joinall([g1,g2,g3])  #可以用JOINALL这样写。

gevent模块提交的方式为异步提交方式。

协程--基于gevent模块的单线程并发

server 端

from gevent import spawn, monkey; monkey.patch_all()
import socket


def communicate(conn):
    """负责通信（相当于服务员，对指定的客人服务）"""
    while True:
        try:
            res = conn.recv(1024)
            if not res:
                break
            conn.send(res.upper())
        except ConnectionResetError:
            break           #这里必须是break否则报错。。

    conn.close()


def server(ip, port):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((ip, port))
    server.listen(5)
    print('listen......')
    while True:
        conn, client = server.accept()   
        #等待连接，有连接连入后执行下面代码，没有连接的话等待连接连入。（相当于接待员）
        print('client IP:%s PORT:%s connection....' % (client[0], client[1]))
        spawn(communicate, conn)
    server.close()


if __name__ == '__main__':
    g = spawn(server, '127.0.0.1', 8927)
    g.join()

client 端

import socket
from threading import Thread, currentThread


def client():
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(('127.0.0.1', 8927))

    while True:
        client.send(('%s hello' % currentThread().name).encode('utf-8'))
        res = client.recv(1024)
        print(res.decode('utf-8'))
    client.close()


if __name__ == '__main__':
    for i in range(500):
        t = Thread(target=client)
        t.start()