1. 程式人生 > >第八篇:python基礎_8 面向對象與網絡編程

第八篇:python基礎_8 面向對象與網絡編程

pro size 賬單 socket 基礎 發生 多態 proc client

本篇內容

  1. 接口與歸一化設計
  2. 多態與多態性
  3. 封裝
  4. 面向對象高級
  5. 異常處理
  6. 網絡編程

一、 接口與歸一化設計

1.定義

(1)歸一化讓使用者無需關心對象的類是什麽,只需要知道這些對象都具備某些功能就可以了,這極大地降低了使用者的使用難度。

(2)歸一化使得高層的外部使用者可以不加區分的處理所有接口兼容的對象集合。

2.模擬接口

#!/usr/bin/env pyhon
#encoding: utf-8
#auth: yanglei

class Interface:    #定義接口Interface類來模仿接口的概念,python中壓根就沒有interface關鍵字來定義一個接口。
    def read(self):     #定接口函數read
        pass

    def write(self):    #定義接口函數write
        pass


class Txt(Interface):   #文本,具體實現read和write
    def read(self):
        print(‘文本數據的讀取方法‘)

    def write(self):
        print(‘文本數據的寫方法‘)

class Sata(Interface):  #磁盤,具體實現read和write
    def du(self):
        print(‘硬盤數據的讀取方法‘)

    def write(self):
        print(‘硬盤數據的寫方法‘)

class Process(Interface):
    def read(self):
        print(‘進程數據的讀取方法‘)

    def xie(self):
        print(‘進程數據的寫方法‘)

t=Txt()
s=Sata()
p=Process()

t.read()
s.read()
p.read()

二、多態與多態性

1.定義

多態:同一種事物的多種形態

多態性:可以在不考慮實例類型的前提下使用實例

2.應用

#!/usr/bin/env pyhon
#encoding: utf-8
#auth: yanglei

class Animal: #同一類事物:動物
    def talk(self):
        pass

class People(Animal): #動物的形態之一:人
    def talk(self):
        print(‘say hello‘)

class Dog(Animal): #動物的形態之二:狗
    def talk(self):
        print(‘say wangwang‘)

class Pig(Animal): #動物的形態之三:豬
    def talk(self):
        print(‘say aoao‘)

class Cat(Animal):
    def talk(self):
        print(‘say miaomiao‘)


class Bird:
    def talk(self):
        print(‘jijiji‘)

#多態性:可以在不考慮實例類型的前提下使用實例
p1=People()
d=Dog()
p2=Pig()
c=Cat()
b=Bird()

# p1.talk()
# d.talk()
# p2.talk()
# c.talk()
# b.talk()

def Talk(animal):
    animal.talk() #p1.talk()

Talk(p1)
Talk(d)
Talk(p2)
Talk(c)
Talk(b)

3.多態性的好處

#!/usr/bin/env pyhon
#encoding: utf-8
#auth: yanglei

#list,str,tuple
l=list([1,2,3])
t=tuple((1,2))
s=str(‘hello‘)

l.__len__()
t.__len__()
s.__len__()

def my_len(obj):
    return obj.__len__()

print(my_len(l))
print(my_len(t))
print(my_len(s))

三、封裝

1.定義

(1)封裝數據屬性:將屬性隱藏起來,然後對外提供訪問屬性的接口,關鍵是我們在接口內定制一些控制邏輯從而嚴格控制使用對數據屬性的使用。

(2)封裝函數屬性:為了隔離復雜度。

2.應用

(1)

#!/usr/bin/env pyhon
#encoding: utf-8
#auth: yanglei

class People:
    def __init__(self,name,age):
        if not isinstance(name,str):
            raise TypeError(‘%s must be str‘ %name)
        if not isinstance(age,int):
            raise TypeError(‘%s must be int‘ %age)
        self.__Name=name
        self.__Age=age
    
    def tell_info(self):
        print(‘<名字:%s 年齡:%s>‘ %(self.__Name,self.__Age))

    def set_info(self,x,y):
        if not isinstance(x,str):
            raise TypeError(‘%s must be str‘ %x)
        if not isinstance(y,int):
            raise TypeError(‘%s must be int‘ %y)
        self.__Name=x
        self.__Age=y

p=People("yanglei",18)
p.tell_info()

p.set_info("Yanglei","19")
p.set_info("Yanglei",19)
p.tell_info()

(2)

(1)取款是功能,而這個功能有很多功能組成:插卡、密碼認證、輸入金額、打印賬單、取錢
(2)對使用者來說,只需要知道取款這個功能即可,其余功能我們都可以隱藏起來,很明顯這麽做
(3)隔離了復雜度,同時也提升了安全性

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class ATM:
    def __card(self):
        print(‘插卡‘)
    def __auth(self):
        print(‘用戶認證‘)
    def __input(self):
        print(‘輸入取款金額‘)
    def __print_bill(self):
        print(‘打印賬單‘)
    def __take_money(self):
        print(‘取款‘)

    def withdraw(self):
        self.__card()
        self.__auth()
        self.__input()
        self.__print_bill()
        self.__take_money()

a=ATM()
a.withdraw()

3.__str__

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class People:
    def __init__(self,name,age,sex):
        self.name=name
        self.age=age
        self.sex=sex

    def __str__(self): #在對象被打印時觸發執行
        return ‘<name:%s age:%s sex:%s>‘ %(self.name,self.age,self.sex)

p1=People(‘egon‘,18,‘male‘)
p2=People(‘alex‘,38,‘male‘)


print(p1)
print(p2)

四、面向對象高級

1.hasattr

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class Foo:
    x=1
    def __init__(self,name):
        self.name=name

    def f1(self):
        print(‘from f1‘)


print(Foo.x) #Foo.__dict__[‘x‘]

f=Foo(‘yanglei‘)
print(f.__dict__)

print(f.name)

print(f.__dict__[‘name‘])

print(hasattr(f,‘name‘)) #f.name
print(hasattr(f,‘f1‘)) #f.f1
print(hasattr(f,‘x‘)) #f.x

2.setattr

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class Foo:
    x=1
    def __init__(self,name):
        self.name=name

    def f1(self):
        print(‘from f1‘)


print(Foo.x) #Foo.__dict__[‘x‘]

f=Foo(‘yanglei‘)

setattr(f,‘age‘,18)#f.age=18x

3.getattr

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class Foo:
    x=1
    def __init__(self,name):
        self.name=name

    def f1(self):
        print(‘from f1‘)


print(Foo.x) #Foo.__dict__[‘x‘]

f=Foo(‘yanglei‘)

print(getattr(f,‘name‘))#f.name
print(getattr(f,‘abc‘,None))#f.abc
print(getattr(f,‘name‘,None))#f.abc

func=getattr(f,‘f1‘)#f.f1
print(func)
func()

4.delattr

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class Foo:
    x=1
    def __init__(self,name):
        self.name=name

    def f1(self):
        print(‘from f1‘)


print(Foo.x) #Foo.__dict__[‘x‘]

f=Foo(‘yanglei‘)

delattr(f,‘name‘)# del f.name
print(f.__dict__)

5.應用

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class Ftpserver:
    def __init__(self,host,port):
        self.host=host
        self.port=port

    def run(self):
        while True:
            cmd=input(‘>>: ‘).strip()
            if not cmd:continue
            if hasattr(self,cmd):
                func=getattr(self,cmd)
                func()
    def get(self):
        print(‘get func‘)

    def put(self):
        print(‘put func‘)

f=Ftpserver(‘192.168.1.2‘,21)
f.run()

五、異常處理

1.定義

異常分為兩種:

(1)語法錯誤

(2)邏輯錯誤

什麽時候用try ...except?
錯誤一定會發生,但是無法預知錯誤發生條件

2.應用

(1)

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

try:
    aaaa
    print(‘==-==>1‘)
    l=[]
    l[3]
    print(‘==-==>2‘)
    d={}
    d[‘x‘]
    print(‘==-==>3‘)
except NameError as e:
    print(e)
except IndexError as e:
    print(e)
except KeyError as e:
    print(e)
except Exception as e:
    print(e)
else:
    print(‘在沒有錯誤的時候執行‘)
finally:
    print(‘無論有無錯誤,都會執行‘)

(2)

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

class YangleiException(BaseException):
    def __init__(self,msg):
        self.msg=msg
    def __str__(self):
        return ‘<%s>‘ %self.msg

raise YangleiException(‘yanglei 的異常‘)

六、網絡編程

1.套接字

服務端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
#買手機
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
#插卡
phone.bind((‘127.0.0.1‘,8080))
#開機
phone.listen(5)
#等電話鏈接
print(‘server start...‘)
conn,client_addr=phone.accept() #(tcp鏈接,client_addr)
print(‘鏈接‘,conn)
print(client_addr)

#基於建立的鏈接,收發消息
client_data=conn.recv(1024)
print(‘客戶端的消息‘,client_data)
conn.send(client_data.upper())

#掛電話鏈接
conn.close()

#關機
phone.close()

客戶端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect((‘127.0.0.1‘,8080))

phone.send(‘hello‘.encode(‘utf-8‘))
server_data=phone.recv(1024)
print(‘服務端回應的消息‘,server_data)

phone.close()

2.通訊循環

服務端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) #就是它,在bind前加
phone.bind((‘127.0.0.1‘,8080))
phone.listen(5)
print(‘server start...‘)
conn,client_addr=phone.accept()

while True: #通訊循環
    client_data=conn.recv(1024)
    # print(‘has rev‘)
    conn.send(client_data.upper())

conn.close()

phone.close()

客戶端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect((‘127.0.0.1‘,8080))

while True:
    msg=input(‘>>: ‘).strip()
    if not msg:continue
    phone.send(msg.encode(‘utf-8‘))
    # print(‘====>has send‘)
    server_data=phone.recv(1024)
    # print(‘====>has recv‘)
    print(server_data.decode(‘utf-8‘))

phone.close()

3.鏈接循環

服務端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) #就是它,在bind前加
phone.bind((‘127.0.0.1‘,8080))
phone.listen(5)
print(‘server start...‘)
while True: #鏈接循環
    conn,client_addr=phone.accept()
    print(conn,client_addr)

    while True: #通訊循環
        try:
            client_data=conn.recv(1024)
            if not client_data:break #針對linux系統
            # print(‘has rev‘)
            conn.send(client_data.upper())
        except Exception: #針對windwos
            break
    conn.close()

phone.close()

客戶端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect((‘127.0.0.1‘,8080))

while True:
    msg=input(‘>>: ‘).strip()
    if not msg:continue
    phone.send(msg.encode(‘utf-8‘))
    # print(‘====>has send‘)
    server_data=phone.recv(1024)
    # print(‘====>has recv‘)
    print(server_data.decode(‘utf-8‘))

phone.close()

4.解決粘包現象

服務端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
import struct
import subprocess
import json

phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) #就是它,在bind前加
phone.bind((‘127.0.0.1‘,8080))
phone.listen(5)
print(‘server start...‘)

while True: #鏈接循環
    conn,client_addr=phone.accept()
    print(conn,client_addr)

    while True: #通訊循環
        try:
            cmd=conn.recv(1024)
            if not cmd:break

            #執行命令,拿到結果
            res=subprocess.Popen(cmd.decode(‘utf-8‘),
                             shell=True,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)

            stdout=res.stdout.read()
            stderr=res.stderr.read()

            #制作報頭
            header_dic={‘total_size‘:len(stdout)+len(stderr),‘md5‘:None}
            header_json=json.dumps(header_dic)
            header_bytes=header_json.encode(‘utf-8‘)

            #1 先發報頭的長度(固定4個bytes)
            conn.send(struct.pack(‘i‘,len(header_bytes)))

            #2 先發報頭
            conn.send(header_bytes)

            #3 再發真實的數據
            conn.send(stdout)
            conn.send(stderr)

        except Exception: #針對windwos
            break
    conn.close()

phone.close()

客戶端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
import struct
import json
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect((‘127.0.0.1‘,8080))

while True:
    cmd=input(‘>>: ‘).strip()
    if not cmd:continue
    #發命令
    phone.send(cmd.encode(‘utf-8‘))

    #先收報頭的長度
    struct_res=phone.recv(4)
    header_size=struct.unpack(‘i‘,struct_res)[0]

    #再收報頭
    header_bytes=phone.recv(header_size)
    head_json=header_bytes.decode(‘utf-8‘)
    head_dic=json.loads(head_json)

    total_size=head_dic[‘total_size‘]
    #再收命令的執行結果
    recv_size=0
    data=b‘‘
    while recv_size < total_size:
        recv_data=phone.recv(1024)
        recv_size+=len(recv_data)
        data+=recv_data

    #打印結果
    print(data.decode(‘gbk‘))

phone.close()

5.實現並發

服務端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socketserver
class MyTcphandler(socketserver.BaseRequestHandler):
    def handle(self):
        while True: #通信循環
            data=self.request.recv(1024)
            self.request.send(data.upper())
if __name__ == ‘__main__‘:
    #取代鏈接循環
    server=socketserver.ThreadingTCPServer((‘127.0.0.1‘,8080),MyTcphandler)
    server.serve_forever()

客戶端:

#!/usr/bin/env pyhon
# encoding: utf-8
# auth: yanglei

import socket
phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect((‘127.0.0.1‘,8080))

while True:
    msg=input(‘>>: ‘).strip()
    if not msg:continue
    phone.send(msg.encode(‘utf-8‘))
    server_data=phone.recv(1024)
    print(server_data.decode(‘utf-8‘))

phone.close()

第八篇:python基礎_8 面向對象與網絡編程