socket编程

客户端与服务端通信

客户端

import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('127.0.0.1', 8000))
while True:re_data = input()client.send(re_data.encode('utf-8'))data = client.recv(1024)print(data.decode('utf-8'))

服务端

import socket, threading
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8000))
server.listen()
def handle_sock(sock, addr):while True:data = sock.recv(1024)print(data.decode('utf-8'))re_data = input()sock.send(re_data.encode('utf-8'))
while True:# 每多一个客户端多一个线程sock, addr = server.accept()threading.Thread(target=handle_sock, args=(sock, addr)).start()

socket实现聊天和多用户连接

多线程的方式，每多一个客户端多一个线程

socket模拟http请求

import socket
from urllib.parse import urlparse
def get_url(url):url = urlparse(url)host = url.netlocpath = url.pathif path == "":path = "/"# 建立socket链接client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect((host, 80))client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode('utf-8'))#二进制类型data = b""while True:d = client.recv(1024)if d:data += delse:breakdata = data.decode('utf-8')print(data)client.close()
if __name__ == '__main__':get_url('https://www.baidu.com/')

多线程、多进程和线程池编程

GIL

同一时刻只有一个线程在CPU上执行字节码，保证线程运行安全，无法将多个线程映射多个CPU上
GIL会根据执行的字节码行数以及时间片释放某线程GIL会给到另外一个线程，GIL在遇到IO操作的时候主动释放，所以导致了下面的代码计算错误

total = 0
def add():global totalfor i in range(1000000):total += 1
def desc():global totalfor i in range(1000000):total -= 1
import threading
thread1 = threading.Thread(target=add)
thread2 = threading.Thread(target=desc)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total)

python多线程

通过Thread类实例化

#对于io操作来说，多线程和多进程性能差别不大
import time
import threading
def get_detail_html(url):print("get detail html started")time.sleep(2)print("get detail html end")
def get_detail_url(url):print("get detail url started")time.sleep(4)print("get detail url end")
if  __name__ == "__main__":thread1 = threading.Thread(target=get_detail_html, args = (url,))thread2 = threading.Thread(target=get_detail_url, args = (url,))start_time = time.time()thread1.start()thread2.start()thread1.join()thread2.join()#当主线程退出的时候， 子线程kill掉print ("last time: {}".format(time.time()-start_time))

通过继承Thread来实现多线程，通过调用父类添加属性，实现run方法。

class GetDetailHtml(threading.Thread):def __init__(self, name):super().__init__(name=name)def run(self):print("get detail html started")time.sleep(2)print("get detail html end")
class GetDetailUrl(threading.Thread):def __init__(self, name):super().__init__(name=name)def run(self):print("get detail url started")time.sleep(4)print("get detail url end")
if  __name__ == "__main__":thread1 = GetDetailHtml("get_detail_html")thread2 = GetDetailUrl("get_detail_url")start_time = time.time()thread1.start()thread2.start()thread1.join()thread2.join()#当主线程退出的时候， 子线程kill掉print ("last time: {}".format(time.time()-start_time))

线程间通信

共享变量,比如detail_url_list=[]，但是在多线程中不安全，主要是在增删过程中

import time
import threading
detail_url_list=[]
#1. 生产者当生产10个url以后就就等待，保证detail_url_list中最多只有十个url
#2. 当url_list为空的时候，消费者就暂停
def get_detail_html():#爬取文章详情页while True:if len(detail_url_list):if len(detail_url_list):url = detail_url_list.pop()print("get detail html started")time.sleep(2)print("get detail html end")else:       time.sleep(1)
def get_detail_url():# 爬取文章列表页while True:print("get detail url started")time.sleep(4)for i in range(20):if len(detail_url_list) >= 10:time.sleep(1)else:     detail_url_list.append("http://projectsedu.com/{id}".format(id=i))print("get detail url end")
if  __name__ == "__main__":thread_detail_url = threading.Thread(target=get_detail_url）thread_detail_url.start()for i in range(10):html_thread = threading.Thread(target=get_detail_html）html_thread.start()

queue 通信，实际上是一个deque()双端队列，是线程安全的

detail_url_queue = Queue(1000)

线程同步

使用锁机制，另只有锁机制的代码段在运行，这样避免内存共享的多线程出现数据错误

死锁：锁没有释放，代码就不运行了

Lock
Rlock：同一个线程里面可以连续多次调用acquire（）,但是与realease（）数量相等
Semaphores：：用于线程数量的锁

# 文件的读写 写一般只用于一个线程,读可以允许有多个
# 做爬虫,控制并发数量,避免反爬虫
import threading, time
class HtmlSpider(threading.Thread):def __init__(self, url, sem):super(HtmlSpider, self).__init__()self.url = urlself.sem = semdef run(self):time.sleep(2)print('get html text success')sem.release()
class UrlProducer(threading.Thread):def __init__(self, sem):super(UrlProducer, self).__init__()self.sem = semdef run(self):for i in range(15):self.sem.acquire()url = 'http://www.baidu.com?p={}'.format(i)html_thread = HtmlSpider(url=url, sem=self.sem)html_thread.start()
if __name__ == '__main__':sem = threading.Semaphore(5)UrlProducer(sem).start()

Conditions：用于复杂的线程通信，使用notify() ，wait()实现对话

class XiaoAi(threading.Thread):def __init__(self, cond):super().__init__(name="小爱")self.cond = conddef run(self):with self.cond:self.cond.wait()print("{} : 在 ".format(self.name))self.cond.notify()self.cond.wait()print("{} : 好啊 ".format(self.name))
class TianMao(threading.Thread):def __init__(self, cond):super().__init__(name="天猫精灵")self.cond = conddef run(self):with self.cond:print("{} : 小爱同学 ".format(self.name))self.cond.notify()self.cond.wait()print("{} : 我们来对古诗吧 ".format(self.name))self.cond.notify()
if __name__ == '__main__':cond = Condition()xiaoai = XiaoAi(cond)tianmao = TianMao(cond)# 启动顺序很重要# 在调用with才能调用notify()# condition有两层锁， 一把底层锁会在线程调用了wait方法的时候释放， 上面的锁会在每次调用wait的时候分配一把并放入到cond的等待队列中，等到notify方法的唤醒xiaoai.start()tianmao.start()

concururrent线程池编码

线程池编码优点

管理线程更加容易,锁的形式管理多线程较为麻烦
主线程中可以获取某一个线程的状态或者某一个任务的状态，以及返回值
当一个线程完成的时候我们主线程能立即知道
futures可以让多线程和多进程编码接口一致

future未来对象：task的返回容器

from concurrent.futures import ThreadPoolExecutor, as_completed,wait
from concurrent.futures import Future
import time
def get_html(times):time.sleep(times)print("get page {} success".format(times))return times
executor = ThreadPoolExecutor(max_workers=3)
# 1. 通过submit函数提交执行的函数到线程池中, submit 是立即返回
task1 = executor.submit(get_html, (3))
# task2 = executor.submit(get_html, (2))
# 3.判断当前执行状态,task为future的对象
# print(task1.done())
# # 执行过程中不可以取消
# print(task2.cancel())
# time.sleep(3)
# print(task1.done())
# # 获取task的返回结果
# print(task1.result())
urls = [3, 2, 4, 3,]
# 任务列表
all_task = [executor.submit(get_html, (url)) for url in urls]
# 2.要获取已经返回成功的task的返回
for future in as_completed(all_task):data = future.result()
#     print('=' * 30, data)
# 2.通过executor获取已经完成的task的返回
# for data in executor.map(get_html,urls):
#     print('=' * 30, data)
# 4.wait 让某些线程阻塞
wait(all_task, return_when='FIRST_COMPLETED')
print('main')

多进程编程—multiprocessing

fork()子进程

import os, time
# fork只能用于linux/unix中
# 线程通过全局变量和queue进行通信,多进程各个进程完全隔离,数据不共享
# 首先运行了父进程后又运行了子进程所以打印了两遍bobby
# 1.新建一个子进程
# pid = os.fork()
# print("bobby")
# if pid == 0:
#     print('子进程 {} ，父进程是： {}.'.format(os.getpid(), os.getppid()))
# else:
#     print('我是父进程：{}.'.format(pid))
# time.sleep(2)

进程：一般对于耗费CPU的操作，多进程多于多线程

import multiprocessing
def get_html(n):time.sleep(n)print('sub_progress success')return n
if __name__ == '__main__':progress = multiprocessing.Process(target=get_html, args=(2,))progress.start()print(progress.pid)# 等待子进程终止后执行一下代码progress.join()print('main progress end)

进程池实现方式一：方法与ThreadPoolExecutor类似

from concurrent.futures import ProcessPoolExecutor

进程池实现方式二：

pool = multiprocessing.Pool(multiprocessing.cpu_count())
# 类似于线程中的future对像
result = pool.apply_async(get_html, args=(1,))
# 将进程池关闭,不再接受新的任务进入
pool.close()
# 主进程等待所有任务完成
pool.join()
print(result.get())
4.imap获取返回值方法和多线程中map类似
for result in pool.imap(get_html,[1,2,3]):print(result)

进程间通信

注意区别多进程中使用的Queue与多线程中的Queue来源不同

from multiprocessing import Queue

pool进程池中不能用multiprocessing中的Queue,需使用Manager中的Queue

from multiprocessing import Manager.Queue

pipe进行多进程通信,只能适用于两个进程,性能高于queue

from multiprocessing import Queue, Manager, Pipe

共享内存

Manager().dict()

协程与异步IO

并发、并行、同步、异步、阻塞、非阻塞

并发：一段时间内有多个程勋在运行，但是任意时刻只有一个程序在运行
并行：在任意时刻有多个程序在运行
同步：在IO操作中，需等待IO操作结果返回才执行其余代码
异步：在IO操作中，不需等待结果就立即执行其余代码（有future，类似消息通信的机制）
阻塞：调用函数时当前线程被挂起
非阻塞：调用函数时不会被挂起，立即返回

C10k（client 10000）问题与IO多路复用（select poll epoll）

阻塞式IO：等待网络返回执行下一步，这个过程CPU空闲，浪费性能。
非阻塞IO：不用等待网络返回，执行下一步
IO复用：select pool epoll,调用这些方法后操作系统会给我们返回，哪些数据报备好，可读；可监听成百上千的socket请求。
异步IO：性能与IO复用差不多

select（IO多路复用） + 回调+事件循环

select poll epoll,一个循环可以监视多个描述符，一旦某个描述符就绪就能够通知程序进行相应的读写操作
select本身是不支持register模式
socket状态变化以后的回调是由程序员完成的
事件循环,不停地请求socket的状态并调用对应的回调函数
回调+事件循环+select(poll\epoll)

import socket, time, queue
from urllib.parse import urlparse
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE
selector = DefaultSelector()
queue = queue.Queue(100)
# 1.io多路复用,使用select完成http请求量
class Fetcher:def connected(self, key):# 注销selector监控的对象selector.unregister(key.fd)# 调用connected函数就说明是一个就绪状态了,所以不需要循环self.client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(self.path, self.host).encode("utf8"))selector.register(self.client.fileno(), events=EVENT_READ, data=self.readable)def readable(self, key):d = self.client.recv(1024)if d:self.data += dprint('=' * 30)else:selector.unregister(key.fd)data = self.data.decode("utf8")html_data = data.split("\r\n\r\n")[1]print(html_data)self.client.close()queue.get(self.spider_url)def get_url(self, url):self.spider_url = urlurl = urlparse(url)self.data = b""self.host = url.netlocself.path = url.pathif self.path == "":self.path = "/"# 建立socket连接self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 非阻塞式ioself.client.setblocking(False)try:self.client.connect((self.host, 80))  # 阻塞不会消耗cpuexcept BlockingIOError as e:pass# 注册(socket对象描述符,写事件,回调函数)selector.register(self.client.fileno(), events=EVENT_WRITE, data=self.connected)
def loop():while not queue.empty():ready = selector.select()for key, mask in ready:call_back = key.datacall_back(key)

回调难度

并不是自上而下的编写，但select是循环加回调的模式，增大了代码编写难度，异常处理困难，共享状态困难

C10M（client 1000000）问题和协程

协程：

同步的方式编写异步代码
采用单线程切换任务
线程是由操作系统来切换，单线程意味着由我们程序员自己编写调度任务
不需要锁（锁会降低并发，线程切换也消耗性能），并发性更高
意味着协程可以暂停函数，并可以在暂停的函数传入值

生成器的send与yield from

yield from会在调用方main与子生成器gen之间建立一个双向通道

#main 调用方 g1(委托生成器) gen 子生成器
def g1(gen):yield from gen
def main():g = g1()g.send(None)

yield form例子

预激middle,建立yield from sales_sum(key)双向通道

final_result = {}
def sales_sum(pro_name):total = 0nums = []while True:x = yieldprint(pro_name + "销量: ", x)if not x:breaktotal += xnums.append(x)return total, nums
def middle(key):while True:# 接受sales_sum return返回值final_result[key] = yield from sales_sum(key)print(key + "销量统计完成！！.")
def main():data_sets = {"bobby牌面膜": [1200, 1500, 3000],"bobby牌手机": [28, 55, 98, 108],"bobby牌大衣": [280, 560, 778, 70],}for key, data_set in data_sets.items():print("start key:", key)# middle 为生成器m = middle(key)# 预激middle协程m.send(None)for value in data_set:向暂停的函数传入值m.send(value)  # 给协程传递每一组的值#生成器继续向下执行m.send(None)print("final_result:", final_result)
if __name__ == '__main__':main()

生成器如何变成协程

python 为了将语义变得更加明确，引入了async和await关键字用于定义原生协程

# 生成器是可以暂停的函数
import inspect
# def gen_func():
#     value=yield from
#     #第一返回值给调用方， 第二调用方通过send方式返回值给gen
#     return "bobby"
# 1. 用同步的方式编写异步的代码， 在适当的时候暂停函数并在适当的时候启动函数
import socket
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE
selector = DefaultSelector()
def get_socket_data():yield "bobby"
def downloader(url):client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.setblocking(False)try:client.connect((host, 80))  # 阻塞不会消耗cpuexcept BlockingIOError as e:passselector.register(self.client.fileno(), EVENT_WRITE, self.connected)source = yield from get_socket_data()data = source.decode("utf8")html_data = data.split("\r\n\r\n")[1]print(html_data)
def download_html(html):html = yield from downloader()
if __name__ == "__main__":# 协程的调度依然是 事件循环+协程模式 ，协程是单线程模式pass

async和await原生协程

await 相当于yield from
await只能接受awaitable对象
用同步的方式编写异步的代码，在适当的时候暂停函数并在适当的时候启动函数

async def downloader(url):return "bobby"
async def download_url(url):# dosomethinghtml = await downloader(url)return html
if __name__ == "__main__":coro = download_url("http://www.imooc.com")# 预激download_url协程coro.send(None)# 与子生成器通信,传递值给xcoro.send(5)

asyncio并发编程

python用于解决异步IO编程的一整套方案

包含特定系统实现的模块化事件
传输与协议抽象
对TCP,UDP,SSL,子进程、延时调用以及其他的集体支持
模仿futures模块但适用于时间循环的Future类
基于yield from 的协议好人任务，可以顺序的编写代码
必须使用一个将产生阻塞的IO的调用，有接口可以把这个事件转移到线程池

事件循环

使用asyncio

import asyncio
import time
async def get_html(url):print("start get url")# 不能用time,sleep(2),因为是单线程模式,同步的形式;若使用await会立即返回一个future对象,达到异步的效果await asyncio.sleep(2)print("end get url")
if __name__ == "__main__":start_time = time.time()loop = asyncio.get_event_loop()tasks = [get_html("http://www.imooc.com") for i in range(100)]# 类似于多线程中的join方法loop.run_until_complete(asyncio.wait(tasks))print(time.time() - start_time)

获得future对象

    # 2.1 通过future获得返回值# get_future = asyncio.ensure_future(get_html("http://www.imooc.com"))# loop.run_until_complete(get_future)# print(get_future.result())# 2.2 task为future的一个子类,获取返回值task = loop.create_task(get_html("http://www.imooc.com"))# 2.2.1协程执行完成后执行回调函数# task.add_done_callback(callback)# 2.2.2若回调需要传递参数,functools.partial偏函数包装函数task.add_done_callback(partial(callback, 'www.baidu.com'))loop.run_until_complete(task)print(task.result())

run_forever(循环一致运行)与run_until_complete(运行完指定协程，停止运行)
task的取消

async def get_html(sleep_times):print("waiting")await asyncio.sleep(sleep_times)print("done after {}s".format(sleep_times))
if __name__ == "__main__":task1 = get_html(2)task2 = get_html(3)task3 = get_html(3)tasks = [task1, task2, task3]loop = asyncio.get_event_loop()try:loop.run_until_complete(asyncio.wait(tasks))except KeyboardInterrupt as e:all_tasks = asyncio.Task.all_tasks()for task in all_tasks:print("cancel task")取消任务print(task.cancel())loop.stop()取消后再次调用loop.run_forever()finally:loop.close()

协程嵌套

call_soon、call_later、call_at、call_soon_threadsafe

import asyncio
def callback(sleep_times, loop):print("success time {}".format(loop.time()))
def stoploop(loop):loop.stop()
# call_later, call_at
if __name__ == "__main__":loop = asyncio.get_event_loop()now = loop.time()loop.call_at(now + 2, callback, 2, loop)loop.call_at(now + 1, callback, 1, loop)loop.call_at(now + 3, callback, 3, loop)# loop.call_soon(stoploop, loop)loop.call_soon(callback, 4, loop)loop.run_forever()

ThreadPoolExecutor+asyncio

场景：协程是单线程的但是遇见阻塞IO怎么处理呢？使用多线程在协程中集成阻塞IO

import asyncio
from concurrent.futures import ThreadPoolExecutor
import socket
from urllib.parse import urlparse
def get_url(url):# 通过socket请求htmlurl = urlparse(url)host = url.netlocpath = url.pathif path == "":path = "/"# 建立socket连接client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# client.setblocking(False)client.connect((host, 80))  # 阻塞不会消耗cpu# 不停的询问连接是否建立好， 需要while循环不停的去检查状态# 做计算任务或者再次发起其他的连接请求client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))data = b""while True:d = client.recv(1024)if d:data += delse:breakdata = data.decode("utf8")html_data = data.split("\r\n\r\n")[1]print(html_data)client.close()
if __name__ == "__main__":import timestart_time = time.time()loop = asyncio.get_event_loop()executor = ThreadPoolExecutor(10)tasks = []for url in range(20):url = "http://shop.projectsedu.com/goods/{}/".format(url)# 将阻塞的代码放到线程池运行task = loop.run_in_executor(executor, get_url, url)tasks.append(task)loop.run_until_complete(asyncio.wait(tasks))print("last time:{}".format(time.time() - start_time))

asyncio模拟http请求

asyncio 没有提供http协议的接口 aiohttp
reader, writer = await asyncio.open_connection(host, 80)

import socket
from urllib.parse import urlparse
async def get_url(url):# 通过socket请求htmlurl = urlparse(url)host = url.netlocpath = url.pathif path == "":path = "/"# 建立socket连接,这步帮助完成了selector.register(),selector.unregister(key.fd)#open_connection(host, 80)为协程，源码最终是把此联结在线程池中运行reader, writer = await asyncio.open_connection(host, 80)writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))all_lines = []async for raw_line in reader:data = raw_line.decode("utf8")all_lines.append(data)html = "\n".join(all_lines)return html
async def main():tasks = []for url in range(200):url = "http://shop.projectsedu.com/goods/{}/".format(url)tasks.append(asyncio.ensure_future(get_url(url)))for task in asyncio.as_completed(tasks):result = await taskprint(result)
if __name__ == "__main__":import timestart_time = time.time()loop = asyncio.get_event_loop()loop.run_until_complete(main())print('last time:{}'.format(time.time() - start_time))

future和task

future为结果容器
task 预激协程，获取StopIteration异常值

asyncio同步和通信

同步：当有两个协程需要调用同样地第三个子协程时，可以用锁的机制将子协程的资源存在缓存当中，避免多次请求重复的资源导致性能浪费
队列：
在协程中Queue相比[]只是多了限流的作用，由于是单线程不考虑共享内存导致计算错误

from aysncio import Lock,Queue

python高级编程相关推荐

python队列线程池_实例详解：python高级编程之消息队列(Queue)与进程池(Pool)
今天为大家带来的内容是:python高级编程之消息队列(Queue)与进程池(Pool),结合了实例的形式详细分析了Python消息队列与进程池的相关原理.使用技巧与操作注意事项!!! Queue消息 ...
python高级编程装饰器_Python装饰器
def my_decorator(function): def _my_decorator(*args, **kw): #在调用实际函数之前做些填充工作 res = function(*args, * ...
python数据符号函数等一切皆对象_第一章：Python高级编程-Python一切皆对象
第一章:Python高级编程-Python一切皆对象 Python3高级核心技术97讲笔记 1. Python一切皆对象 1.1 函数和类也是对象,属于Python的一等公民 "" ...
【Python高级编程】
Python高级编程:技巧代码的玄学与艺术一.编程语言介绍 Python 作为一门优秀的编程语言,有着很多优势: 简单易学 Python有简单的语法,易于阅读和学习,很适合初学者.它的设计哲学是&q ...
python高级编程-网络编程、多任务
python高级编程 1 IP地址用来在网络中标记一台电脑:在本地局域网上是唯一的. 2 端口一个程序需要收发网络数据,就需要端口号. 3 socket 创建socket # 创建tcp sock ...
Python 高级编程笔记之类别
目录: 子类化内建类型访问超类中的方法-super 描述符 & 属性元编程主要内容: 1.子类化内建类型 # -*- coding:utf-8 -*-class Folder(list) ...
python高级编程(3) - 深入类和多态
深入类和多态一,鸭子类型和多态鸭子类型当你看到一只鸟走起来想鸭子,游泳起来像鸭子,叫起来像鸭子,那么这只鸟就叫做鸭子类型我们并不关心对象是什么类型,到底是不是鸭子,只关心行为. 我们只关心一个 ...
4万字【Python高级编程】保姆式教学，进阶感觉到吃力？学完这些就轻松了
前几天和一个小伙子聊天时,发现了一个问题,他从零开始学Python,学完列表.字典和函数等基础之后,就开始往爬虫方向进阶学习,结果又花了一个多月的时间,啥也没学成,反而寸步难行. 其实这个问题的主要原 ...
python高级编程函数_Python高级编程之十大装B语法
for - else 什么?不是 if 和 else 才是原配吗?No,你可能不知道,else 是个脚踩两只船的家伙,for 和 else 也是一对,而且是合法的.十大装B语法,for-else 绝对 ...
python高级编程之网络编程
Python高级之网络编程端口端口分类知名端口动态端口查看端口 socket简介电脑上进程之间的通信什么是socket 创建socket 使用UDP套接字发送数据使用UDP套接字接受数 ...

python高级编程

socket编程

客户端与服务端通信

socket实现聊天和多用户连接

socket模拟http请求

多线程、多进程和线程池编程

GIL

python多线程

线程间通信

线程同步

concururrent线程池编码

多进程编程—multiprocessing

进程间通信

协程与异步IO

并发、并行、同步、异步、阻塞、非阻塞

C10k（client 10000）问题与IO多路复用（select poll epoll）

select（IO多路复用） + 回调+事件循环

回调难度

C10M（client 1000000）问题和协程

生成器的send与yield from

生成器如何变成协程

async和await原生协程

asyncio并发编程

事件循环

协程嵌套

call_soon、call_later、call_at、call_soon_threadsafe

ThreadPoolExecutor+asyncio

asyncio模拟http请求

future和task

asyncio同步和通信

python高级编程相关推荐

最新文章

热门文章