python中socket进程线程协程池的创建方式

Posted tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python中socket进程线程协程池的创建方式相关的知识,希望对你有一定的参考价值。

一、TCP-socket
服务端:
import socket
tcp_sk = socket.socket()
tcp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
tcp_sk.bind((127.0.0.1,8000))
tcp_sk.listen()

conn,addr = tcp_sk.accept()

conn.send(你好.encode(utf-8))
print(conn.recv(1024).decode(utf-8))

conn.close()
tcp_sk.close()

客户端:
import socket
sk = socket.socket()
sk.connect((127.0.0.1,8000))

print(sk.recv(1024).decode(utf-8))
sk.send(嘿嘿嘿.encode(utf-8))

sk.close()


二、UDP-socket
服务端:
import socket
udp_sk = socket.socket(type=socket.SOCK_DGRAM)
udp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
udp_sk.bind((127.0.0.1,8001))

msg,addr = udp_sk.recvfrom(1024)
print(msg.decode(utf-8))

udp_sk.sendto(你好.encode(utf-8),addr)

udp_sk.close()


客户端:
import socket
sk = socket.socket(type=socket.SOCK_DGRAM)

sk.sendto(哈哈.encode(utf-8),(127.0.0.1,8001))

msg,addr = sk.recvfrom(1024)
print(msg.decode(utf-8))
sk.close()





三、socketserver
服务端:
import socketserver

class Myserver(socketserver.BaseRequestHandler):
    def handle(self):
        conn = self.request
        while True:
            conn.send(bhello)
            print(conn.recv(1024).decode(utf-8))

socketserver.TCPServer.allow_reuse_address = True

server = socketserver.ThreadingTCPServer((127.0.0.1,8080),Myserver)
server.serve_forever()



客户端:
import socket
sk = socket.socket()
sk.connect((127.0.0.1,8080))
while True:
    ret = sk.recv(1024)
    print(ret.decode(utf-8))
    sk.send(bhiworld)
sk.close()


四、进程
方式一、
from multiprocessing import Process
def func(arg):
    print(arg)

if __name__ == __main__:
    p = Process(target=func,args=(子进程,))
    p.start()
    p.join()
    print(主进程)
    
方式二、
from multiprocessing import Process

class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name

    def run(self):
        print(self.name)

if __name__ == __main__:
    p = MyProcess(小明)
    p.start()


五、线程
方式一、
from threading import Thread
import time

def sleep_boy(name):
    time.sleep(1)
    print(%s is sleeping %name)

t = Thread(target=sleep_boy,args=(xiaoming,))  # 这里可以不需要main,因为现在只是在一个进程内操作,不需要导入进程就不会import主进程了
t.start()
print(主线程)


方式二、
from threading import Thread
import time

class Sleep_boy(Thread):
    def __init__(self,name):
        super().__init__()
        self.name = name

    def run(self):
        time.sleep(1)
        print(%s is sleeping % self.name)

t = Sleep_boy(xiaoming)
t.start()
print(主线程)



六、协程
1、greenlet例子:
import time
from greenlet import greenlet
def cooking():
    print(cooking 1)
    g2.switch()      # 切换到g2,让g2的函数工作
    time.sleep(1)
    print(cooking 2)

def watch():
    print(watch TV 1)
    time.sleep(1)
    print(watch TV 2)
    g1.switch()    # 切换到g1,让g1的函数工作

g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch()        # 切换到g1,让g1的函数工作

greenlet的缺陷:很显然greenlet实现了协程的切换功能,可以自己设置什么时候切,在哪切,但是它遇到阻塞并没有自动切换,
因此并不能提高效率。所以一般我们都使用gevent模块实现协程


2、gevent例子:
from gevent import monkey
monkey.patch_all()
import time
import gevent

def cooking():
    print(cooking 1)
    time.sleep(1)
    print(cooking 2)

def watch():
    print(watch TV 1)
    time.sleep(1)
    print(watch TV 2)

g1 = gevent.spawn(cooking) # 自动检测阻塞事件,遇见阻塞了就会进行切换
g2 = gevent.spawn(watch)

g1.join()    # 阻塞直到g1结束
g2.join()    # 阻塞直到g2结束


七、进程池
1、同步提交apply:
import os
import time
from multiprocessing import Pool

def test(num):
    time.sleep(1)
    print(%s:%s %(num,os.getpid()))
    return num*2

if __name__ == __main__:
    p = Pool()
    for i in range(20):
        res = p.apply(test,args=(i,))  # 提交任务的方法 同步提交
        print(-->,res)                        # res就是test的return的值,同步提交的返回值可以直接使用

        
2、异步提交apply_async:
2-1无返回值:
import time
from multiprocessing import Pool

def func(num):
    time.sleep(1)
    print(做了%s件衣服%num)

if __name__ == __main__:
    p = Pool(4)  # 进程池中创建4个进程,不写的话,默认值为你电脑的CUP数量
    for i in range(50):
        p.apply_async(func,args=(i,)) # 异步提交func到一个子进程中执行,没有返回值的情况
    p.close() # 关闭进程池,用户不能再向这个池中提交任务了
    p.join()   # 阻塞,直到进程池中所有的任务都被执行完        


2-2有返回值:
import time
import os
from multiprocessing import Pool

def test(num):
    time.sleep(1)
    print(%s:%s %(num,os.getpid()))
    return num*2

if __name__ == __main__:
    p = Pool()
    res_lst = []
    for i in range(20):
        res = p.apply_async(test,args=(i,))   # 提交任务的方法 异步提交
        res_lst.append(res)
    for res in res_lst:
        print(res.get())  # 异步提交的返回值需要get,get有阻塞效果,此时就不需要close和join
        


2-3map:
map接收一个函数和一个可迭代对象,是异步提交的简化版本,自带close和join方法
可迭代对象的每一个值就是函数接收的实参,可迭代对象的长度就是创建的任务数量
map可以直接拿到返回值的可迭代对象(列表),循环就可以获取返回值

import time
from multiprocessing import Pool

def func(num):
    print(子进程:,num)
    # time.sleep(1)
    return num

if __name__ == __main__:
    p = Pool()
    ret = p.map(func,range(10))   # ret是列表
    for i in ret:
        print(返回值:,i)    
        
        
        
        
2-4回调函数:
import os
from multiprocessing import Pool

def func(i):
    print(子进程:,os.getpid())
    return i

def call_back(res):
    print(回调函数:,os.getpid())
    print(res--->,res)

if __name__ == __main__:
    p = Pool()
    print(主进程:,os.getpid())
    p.apply_async(func,args=(1,),callback=call_back)  # callback关键字传参,参数是回调函数
    p.close()
    p.join()

    

八、进程池、线程池
线程池:
1import time
from concurrent.futures import ThreadPoolExecutor

def func(i):
    print(thread,i)
    time.sleep(1)
    print(thread %s end%i)

tp = ThreadPoolExecutor(5)  # 相当于tp = Pool(5)
tp.submit(func,1)           # 相当于tp.apply_async(func,args=(1,))
tp.shutdown()               # 相当于tp.close()  +  tp.join()
print(主线程)


2import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread

def func(i):
    print(thread,i,currentThread().ident)
    time.sleep(1)
    print(thread %s end%i)

tp = ThreadPoolExecutor(5)
for i in range(20):
    tp.submit(func,i) 
tp.shutdown()  # shutdown一次就够了,会自动把所有的线程都join()
print(主线程)


3、返回值
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread

def func(i):
    print(thread,i,currentThread().ident)
    time.sleep(1)
    print(thread %s end %i)
    return i * *

tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
    ret = tp.submit(func,i)
    ret_lst.append(ret)

for ret in ret_lst:
    print(ret.result())  # 相当于ret.get()

print(主线程)



4、map
map接收一个函数和一个可迭代对象
可迭代对象的每一个值就是函数接收的实参,可迭代对象的长度就是创建的线程数量
map可以直接拿到返回值的可迭代对象(列表),循环就可以获取返回值

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(thread,i)
    time.sleep(1)
    print(thread %s end%i)
    return i * *

tp = ThreadPoolExecutor(5)
ret = tp.map(func,range(20))
for i in ret:
    print(i)


5、回调函数
    回调函数在进程池是由主进程实现的
    回调函数在线程池是由子线程实现的


import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread

def func(i):
    print(thread,i,currentThread().ident)
    time.sleep(1)
    print(thread %s end%i)
    return i * *

def call_back(arg):
    print(call back : ,currentThread().ident)
    print(ret : ,arg.result())  # multiprocessing的Pool回调函数中的参数不需要get(),这里需要result()

tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
    tp.submit(func,i).add_done_callback(call_back)  # 使用add_done_callback()方法实现回调函数 
print(主线程,currentThread().ident)

 

以上是关于python中socket进程线程协程池的创建方式的主要内容,如果未能解决你的问题,请参考以下文章

Go语言 | 协程池的应用(可能是全网最适合小白的教程)

Go语言 | 协程池的应用(可能是全网最适合小白的教程)

白话 Golang 协程池

白话 Golang 协程池

Go简单实现协程池

Python:线程进程与协程——线程池