python IO编程、线程和进程、网络编程

1.Socket类型

套接字格式为:socket(family,type[,protocal]]),使用给定的地址族，套接字类型、协议编号来创建套接字

socket.AF_UNIX 只能够用于单一的Unix系统进程间通信

socket.AF_INET 服务器之间网络通信

socket.AF_INNET6 IPv6

socket.SOCKET_STREAM 流式socket,用于TCP

socket.SOCK_RAW 原始套接字，普通的套接字无法理解ICMP,IGMP等网络报文，而SOCK_RAW可以，其次，SOCK_RAW也可以理解特殊的

Ipv4报文,利用原始套接字可以通过IP_HDRINCL套接字选项由用户构造IP头

创建TCP Socket s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)

创建UDP Socket s=socket.socket(socket.AF_INET,socket.SOck_DGRAM)

常用变成函数

服务端Socket函数 s.bind(address) 将套接字绑定到地址，在AF_INET下，以元组(host,port)的形式表示地址

s.listen(backlog) 开始监听TCP传入连接。backlog指定在拒绝连接之前，操作系统可以挂起的最大连接数量。该值至少为1，大部分应用程序设置为5

s.accept() 接受TCP连接并返回(conn,address),其中conn是新的套接字对象，可以用来接收和发送消息，address是连接客户端的地址

客户端Socket函数

s.connect(address) 连接到address处的套接字。一般address的格式为元组(hostname,port) ,如果连接出错，返回socket.error错误

s.connect_ext(address) 功能与connect(address)相同，但成功返回0，失败返回error的值

公共Socket函数

s.recv(buflen,flage) :接受TCP套接字的数据，数据以字符形式返回，buflen指定要接收的最大数据量,flage提供有关消息的其他信息

s.send(string,flage) 发送TCP数据，将string中的数据发送到连接的套接字，返回值是要发送的字节数，该数量可能小于string的字节大小

s.sendall(string，flage) 完整发送TCP数据，将string中的数据发送到连接的套接字，但在返回之前会尝试发送所有数据，成功返回None,失败抛出一哦下行

s.recvfrom(buflen,flage) 接受UDP套接字的数据，与recv()类似，但返回值是(data,address).其中data 是包含接受数据的字符串，address是发送数据的套接字地址

s.sendto(string,flage,address) 发送UDP数据。将数据发送到套接字，address是形式为(ipadder,port)的元组，指定远程地址，返回值是发送的字节数

s.close() 关闭套接字

s.getpeername() 返回连接套接字的远程地址。返回值通常是元组(ipadder,port)

s.getsockname() 返回套接字自己的地址，通常是一个元组(ipadder,port)

s.setsockopt(level,optname,value) 设置给定套接字选项的值

s.getsockopt() 返回套接字选项的值

s.settimeout(timeout) 设置套接字操作的超时时间，timeout是一个浮点数,单位是秒。值为None表示没有超时，一般超时应该在创建链接字

时设置，因为热他们可能会用于连接操作

s,setblocking(flage) 如果flage值为0，则将套接字设置为非阻塞模式，否则将设置为阻塞模式，默认是阻塞模式，在非阻塞模式下，如果调用方

recv()没有发现任何数据，或send()调用无法立即发送数据，将引起socket.error异常

TCP编程:面向连接的编程方式消息可靠

服务端程序编写

1)创建Socket,绑定Socket到本地IP与端口

2)开始监听连接

3)进入循环，不断接收客户端的连接请求

4)接收传来的数据，并发送给对方数据

5)传输完成关闭Socket

import socket
import threading
import time


# 第四步：接收传来的数据，并发送给对方数据
def dealClient(sock, addr):
    print('Accept new connection from %s:%s.....' % addr)
    sock.send(b'Hello, I am server!')
    while True:
        data = sock.recv(1024)
        time.sleep(1)
        if not data or data.decode('utf-8') == 'exit':
            break
        print('----->>%s!' % data.decode('utf-8'))
        sock.send(b'Loop Msg: %s!' % data.decode('utf-8').encode('utf-8'))  # 只能传送bytes
    # 第五部关闭Socket
    sock.close()
    print('Connection from %s:%s closed.' % addr)


if __name__ == '__main__':
    # 第一步：创建一个基于IPv4和TCP协议的Socket
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('127.0.0.1', 9999))
    # 第二步：监听
    s.listen(5)
    print('Waiting for connection......')
    while True:
        print('-----------------')
        # 第三步:接收一个新连接:
        sock, addr = s.accept()  # 等待一个链接的到来，会一直等待
        # 创建新线程来处理TCP连接
        t = threading.Thread(target=dealClient, args=(sock, addr))
        t.start()

View Code

客户端程序编写

1)创建Socket,连接远端地址

2)连接后发送数据和接收数据

3)传输完毕后，关闭Socket

import socket
import  time
import  random
#初始化Socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#连接目标IP和端口
s.connect(('127.0.0.1',9999))
time.sleep(random.random())
#接收消息
print('---->>',s.recv(1024).decode('utf-8'))#没有消息就会一直等待，或者知道进程结束
#发送消息
s.send(b'Hello,I am a client1')
print('--->>',s.recv(1024).decode('utf-8'))
s.send(b'exit')
#关闭Socket
s.close()

View Code

UDP编程

服务器端

1)创建Socket, 绑定指定的IP和端口

2)直接发送数据和接收数据

3)关闭Socket

import socket

# UDp服务进程

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(('127.0.0.1', 9999))
print('Bind UDP om 9999....')
while True:
    data, addr = s.recvfrom(1024)
    print('Message is %s' % data.decode('utf-8'))
    print(' Received from %s:%s' % addr)
    s.sendto(b'Hello,%s!' % data, addr)

View Code

客户端

直接创建Socket 然后与服务器进行数据交换

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
for data in [b'Hello', b'World']:
    s.sendto(data, ('127.0.0.1', 9999))
    print(s.recv(1024).decode('utf-8'))
s.close()

View Code

分布式简单案例：其实就是把本地Queue数据网络化，然后一个进程产生taskQueue,另一个处理taskQueue任务并把结果甚至到resultQueue中。

分布式进程指的是将Process进程分布到多台机器上，充分利用多台机器的性能完成复杂的任务：当做爬虫时，遇到抓取某个网站的所有图片问题

可以通过设置一个进程抓取所有的连接放在Queue中，另一个进程从Queue中下载所有的图片数据，再保存到resultQueue中？？如何实现Queue共享

分布式过程

(1)建立队列Queue,用来进行进程间的通信，服务进程创建队列task_queue,用来作为传递任务给任务进程的通道；服务进程创建队列result_queue,

作为任务进程任务完成后回复服务进程，可以通过Queuemanager获得Queue接口来添加任务。

2)把第一步建立的队列在网络上注册，暴露给其他进程，注册后获得网络队列，相当于本地队列的映像

3)建立一个对象(Queuemanager(BaseManager))实例manager,绑定端口和验证口令。

4)启动第三步中建立的实例，即启动管理manager,监管信息通道

5)通过管理实例的方法获得通过网络访问的Queue对象，即再把网络队列实体化成可以使用的本地队列

6)创建任务到‘本地’队列中，自动上传任务到网络队列中，分配给任务进程处理

服务器进程代码

# Linux版本的服务器代码


import random, time, queue

from multiprocessing.managers import BaseManager


class Queuemanager(BaseManager):
    pass
# 第一步：建立task_queue 和result_queue,用来存放任务和结果
result_queue = queue.Queue()
task_queue = queue.Queue()
# 第二步:将Queue对象在网络中暴露
Queuemanager.register('get_task_queue', callable=lambda: task_queue)
Queuemanager.register('get_result_queue', callable=lambda: result_queue)
# 第三步:绑定端口8001，设置验证口令'qiye'
manager = Queuemanager(address=('', 8001), authkey='qiye'.encode())
# 第四步:启动管理，监听信息通道
manager.start()
# 第五步:通过管理实例的方法获得通过网络访问的Queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()
# 第六步：添加任务
for url in ['ImageUrl_' + i for i in range(10)]:
    print('put task %s...' % url)
    task.put(url)
# 返回的结果
print('try get result.....')
for i in range(10):
    print('result is %s' % result.get(timeout=10))
manager.shutdown()

View Code

# Windows下的服务器代码
import queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support

task_number = 10
task_queue = queue.Queue(task_number)
result_queue = queue.Queue(task_number)


def get_task():
    return task_queue


def get_result():
    return result_queue


class QueueManager(BaseManager):
    pass


def win_run():
    QueueManager.register('get_result_queue', callable=get_result)
    QueueManager.register('get_task_queue', callable=get_task)
    # 绑定端口并设置验证
    manager = QueueManager(address=('127.0.0.1', 8003), authkey='qiye'.encode())
    manager.start()
    try:
        task = manager.get_task_queue()
        result = manager.get_result_queue()
        for url in ['ImageUrl_%d' % (i) for i in range(task_number)]:
            print('put task %s...' % url)
            task.put(url)
        print('try get result....')
        for i in range(task_number):
            print('result is %s' % result.get(timeout=100))
    except Exception as e:
        print(e)
    finally:
        manager.shutdown()
if __name__ == '__main__':
    freeze_support();
    win_run()

View Code

任务进程

1)使用QueueManager注册用于获取Queue的方法名称，任务进程只能通过名称在网络上获取Queue

2)连接服务器，端口和验证口令注意保持与服务器进程中完全一致

3)从网络上获取Queue,进行本地化

4)从task队列获取任务，并把结果写入result队列。

# 任务进程
import time
from multiprocessing.managers import BaseManager
# 创建类似的QueueManager
class QueueManager(BaseManager):
    pass
# 第一步：使用QueueManager注册用于获取Queue的方法名称
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
# 连接服务器
server_addr = '127.0.0.1'
print('Connect to server %s.....' % server_addr)
# 端口和验证码口令注意保持与服务器进程完全一致
m = QueueManager(address=(server_addr, 8003), authkey='qiye'.encode())
# 网络链接
m.connect()
# 第三步：获取Queue的对象
task = m.get_task_queue()
result = m.get_result_queue()
# 第四步：从task队列获取任务，并把结果写入result队列
while (not task.empty()):
    image_url = task.get(True, timeout=5)
    print('run task downlaod%s....' % image_url)
    time.sleep(1)
    result.put('%s----->success' % image_url)
print('work exit.')

View Code