python网络编程（进程与多线程）

from multiprocessing import Process
import os
import time
def info(name): print("name:",name)
print('parent process:', os.getppid())
print('process id:', os.getpid())
print("------------------")
time.sleep(1)def foo(name): info(name)if __name__ == '__main__': info('main process line') p1 = Process(target=info, args=('alvin',))
p2 = Process(target=foo, args=('egon',))
p1.start()
p2.start() p1.join()
p2.join() print("ending")
运行结果：name: main process line
parent process: 7904#pycharm的进程pid
process id: 11424#这个是python解释器的pid
------------------
name: alvin
parent process: 11424
process id: 9628
------------------
name: egon
parent process: 11424
process id: 9276
------------------
ending

from multiprocessing import Process
import time,random
import os
def piao(name):
print(os.getppid(),os.getpid())
print('%s is piaoing' %name)
time.sleep(random.randint(1,3))
print('%s is piao end' %name)
if __name__ == '__main__':
p1=Process(target=piao,kwargs={'name':'alex',})
p2=Process(target=piao,args=('wupeiqi',))
p3=Process(target=piao,kwargs={'name':'yuanhao',})
p1.start()
p2.start()
p3.start()
print('主进程',os.getpid())
#os.getppid(),os.getpid()
#父进程id，当前进程id

from multiprocessing import Process
import time,random
import os
class Piao(Process):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
print(os.getppid(),os.getpid())
print('%s is piaoing' %self.name)
# time.sleep(random.randint(1,3))
print('%s is piao end' %self.name)
if __name__ == '__main__':
p1=Piao('alex')
p2=Piao('wupeiqi')
p3=Piao('yuanhao') p1.start()
p2.start()
p3.start()
print('主进程',os.getpid(),os.getppid())

import time"""
传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。
如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。
"""
# 注意到consumer函数是一个generator（生成器）:
# 任何包含yield关键字的函数都会自动成为生成器(generator)对象def consumer():
r = ''
while True:
# 3、consumer通过yield拿到消息，处理，又通过yield把结果传回；
# yield指令具有return关键字的作用。然后函数的堆栈会自动冻结(freeze)在这一行。
# 当函数调用者的下一次利用next()或generator.send()或for-in来再次调用该函数时，
# 就会从yield代码的下一行开始，继续执行，再返回下一次迭代结果。通过这种方式，迭代器可以实现无限序列和惰性求值。
n = yield r
if not n:
return
print('[CONSUMER] ←← Consuming %s...' % n)
time.sleep(1)
r = '200 OK'
def produce(c):
# 1、首先调用c.next()启动生成器
next(c)
n = 0
while n < 5:
n = n + 1
print('[PRODUCER] →→ Producing %s...' % n)
# 2、然后，一旦生产了东西，通过c.send(n)切换到consumer执行；
cr = c.send(n)
# 4、produce拿到consumer处理的结果，继续生产下一条消息；
print('[PRODUCER] Consumer return: %s' % cr)
# 5、produce决定不生产了，通过c.close()关闭consumer，整个过程结束。
c.close()
if __name__=='__main__':
# 6、整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务。
c = consumer()
produce(c)'''
result:[PRODUCER] →→ Producing 1...
[CONSUMER] ←← Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 2...
[CONSUMER] ←← Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 3...
[CONSUMER] ←← Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 4...
[CONSUMER] ←← Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 5...
[CONSUMER] ←← Consuming 5...
[PRODUCER] Consumer return: 200 OK
'''

from greenlet import greenletdef 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()

import gevent
import timedef foo():
print("running in foo")
gevent.sleep(2)
print("switch to foo again")def bar():
print("switch to bar")
gevent.sleep(5)
print("switch to bar again")start=time.time()gevent.joinall(
[gevent.spawn(foo),
gevent.spawn(bar)]
)print(time.time()-start)

from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import timedef f(url):
print('GET: %s' % url)
resp = request.urlopen(url)
data = resp.read()
print('%d bytes received from %s.' % (len(data), url))start=time.time()gevent.joinall([
gevent.spawn(f, 'https://itk.org/'),
gevent.spawn(f, 'https://www.github.com/'),
gevent.spawn(f, 'https://zhihu.com/'),
])# f('https://itk.org/')这是分别爬，串行的操作
# f('https://www.github.com/')
# f('https://zhihu.com/')print(time.time()-start)

import socket
import timesock=socket.socket()sock.bind(("127.0.0.1",8800))sock.listen(5)sock.setblocking(False)#设置为非阻塞while 1:
try:
conn,addr=sock.accept() # 阻塞等待链接
except Exception as e:
print(e)
time.sleep(3)

import time
import socket
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)while True:
sk.connect(('127.0.0.1',6667))
print("hello")
sk.sendall(bytes("hello","utf8"))
time.sleep(2)
break

import socket
import timeimport select
sock=socket.socket()sock.bind(("127.0.0.1",8800))sock.listen(5)sock.setblocking(False)
inputs=[sock,]print("sock",sock)while 1:
r,w,e=select.select(inputs,[],[]) # 监听有变化的套接字 inputs=[sock,conn1,conn2,conn3..]
print("r",r)#select就卡在这一有链接来就开始操作，没链接就block阻塞
print("r",w)
print("r",e)
for obj in r: # 第一次 [sock,] 第二次 #[conn1,]
if obj==sock:#sock是用户来连接我我有变化，将新的链接加入
print('change')
conn,addr=obj.accept()
inputs.append(conn) # inputs=[sock,conn] else:#客户端传来消息，那么我的conn发生变化，进行数据交互
data=obj.recv(1024)
print(data.decode("utf8"))
send_data=input(">>>")
obj.send(send_data.encode("utf8"))

import socketsock=socket.socket()sock.connect(("127.0.0.1",8800))while 1: data=input("input>>>")
sock.send(data.encode("utf8"))
rece_data=sock.recv(1024)
print(rece_data.decode("utf8"))sock.close()

import selectors # 基于select模块实现的IO多路复用，建议大家使用import socketsock=socket.socket()
sock.bind(("127.0.0.1",8800))sock.listen(5)sock.setblocking(False)sel=selectors.DefaultSelector() #根据具体平台选择最佳IO多路机制，比如在linux，选择epolldef read(conn,mask): try:
data=conn.recv(1024)
print(data.decode("UTF8"))
data2=input(">>>")
conn.send(data2.encode("utf8"))
except Exception:
sel.unregister(conn)def accept(sock,mask): conn, addr = sock.accept()
print("conn",conn)
sel.register(conn,selectors.EVENT_READ,read)sel.register(sock,selectors.EVENT_READ,accept) # selectors对象注册事件，监听谁就要注册谁，第二个默认，第三个监听对象有变化运行这个函数while 1: print("wating...")
events=sel.select() # 监听 [(key1,mask1),(key2,mask2)]第一次只监听sock有链接过来才会继续
for key,mask in events: # print(key.fileobj) # conn
# print(key.data) # read
func=key.data#accept函数
obj=key.fileobj#sock func(obj,mask) # 1 accept(sock,mask) # 2 read(conn,mask)

import queue#q=queue.Queue(3) # 默认是 先进先出(FIFO)管道容纳最大值# q.put(111)#put塞值
# q.put("hello")
# q.put(222)
#
# q.put(223,False)
#
# print(q.get())#get取值取不到值就是block
# # print(q.get())
# # print(q.get())
# #
# q.get(False)#取不到还把block转成false就会报错了。# queue 优点： 线程安全的

# join和task_done# q=queue.Queue(5)# q.put(111)
# q.put(222)
# q.put(22222)
#
#
# while not q.empty():
# a=q.get()
# print(a)
#q.task_done()#任务完成了告诉一下join# b=q.get()
# print(b)
# q.task_done()# q.join()只有所有的都任务都结束才不block否则都卡住
#
# print("ending")

# 先进后出模式# q=queue.LifoQueue() # Lifo last in first out
#
#
# q.put(111)
# q.put(222)
# q.put(333)
#
# print(q.get())

# 优先级# q=queue.PriorityQueue()
#
# q.put([4,"hello4"])
# q.put([1,"hello"])
# q.put([2,"hello2"])
#
# print(q.get())
# print(q.get())

#生产者消费者模型import time,random
import queue,threadingq = queue.Queue()def Producer(name):
count = 0
while count <10:#生产数据
print("making........")
time.sleep(2)#生产时间
q.put(count)
print('Producer %s has produced %s baozi..' %(name, count)) count +=1
#q.task_done()
#q.join()
print("ok......")def Consumer(name):#消费者
count = 0
while count <10:
time.sleep(1)
if not q.empty():
data = q.get()
#q.task_done()
#q.join()
print(data)
print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
else:
print("-----no baozi anymore----") count +=1p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))