3.4 - 面向对象 - 总结练习题
Encapsulation 封装
在类中对数据的赋值、内部调用对外部用户是透明的，这使类变成了一个胶囊或容器，里面包含着类的数据和方法
Inheritance 继承
一个类可以派生出子类，在这个父类里定义的属性、方法自动被子类继承
Polymorphism 多态
多态是面向对象的重要特性,简单点说:“一个接口，多种实现”，指一个基类中派生出了不同的子类，且每个子类在继承了同样的方法名的同时又对父类的方法做了不同的实现，这就是同一种事物表现出的多种形态。
2.类的属性和对象的属性有什么区别?
类的属性：数据属性和函数属性，数据属性是所有对象共有的，函数属性是绑定给对象使用的
对象的属性：对象是类的实例化
3.面向过程编程与面向对象编程的区别与应用场景?
面向过程：复杂的问题流程化，简单化 应用场景：不在需要扩展了，监控脚本，自动部署脚本，软件解压安装
面向对象：特征与技能的结合体 一切皆对象 应用场景：用户需求经常变化，互联网应用，游戏，企业内部应用
4.类和对象在内存中是如何保存的。
类和对象的属性：以字典的形式保存的。
5.什么是绑定到对象的方法、绑定到类的方法、解除绑定的函数、如何定义，如何调用，给谁用？有什么特性
绑定到对象的方法：就应该由对象来调用，def tell_info(self):... obj.tell_info()
绑定到类的方法：就应该由类来调用，@classmethod def from_conf(cls):... class.from_conf()
非绑定方法：不与类或对象绑定，谁都可以调用，@staticmethod def create_id():... obj.create_if()/class.create_id()
6.使用实例进行 获取、设置、删除 数据, 分别会触发类的什么私有方法
class A(object):
def __setitem__(self, key, value):
print('setitem')
self.__dict__[key] = value
def __getitem__(self, item):
print("触发 __getitem__方法 ")
return self.__dict__.get(item)
def __delitem__(self, key):
print('delitem')
del self.__dict__[key]
a = A()
print(a.__dict__)
a["key"] = "val" #触发__setitem__方法
print(a.__dict__)
c = a["key"] # 触发 __getitem__方法
print(c)
del a["key"] #触发__delitem__方法
print(a.__dict__)
7.python中经典类和新式类的区别
经典类:py2 没有继承object的类,以及它的子类都称之为经典类 -->深度优先
新式类:py3 继承object的类,以及它的子类都称之为新式类 -->广度优先
8.如下示例, 请用面向对象的形式优化以下代码
1 # 1、在没有学习类这个概念时，数据与功能是分离的
2 # def exc1(host,port,db,charset):
3 # conn=connect(host,port,db,charset)
4 # conn.execute(sql)
5 # return xxx
6 # def exc2(host,port,db,charset,proc_name)
7 # conn=connect(host,port,db,charset)
8 # conn.call_proc(sql)
9 # return xxx
10 # # 每次调用都需要重复传入一堆参数
11 # exc1('127.0.0.1',3306,'db1','utf8','select * from tb1;')
12 # exc2('127.0.0.1',3306,'db1','utf8','存储过程的名字')
13
示例1：
import socket
class exc:
host = "127.0.0.1"
port = 3306
db = "db1"
charset = "utf-8"
conn = connect(host,port,db,charset)
def exc1(self,*args):
self.conn.call_proc(sql)
return xxx
def exc2(self,*args):
self.conn.call_proc(sql)
return xxx
ex = exc()
ex.exc1("select * from tb1;")
ex.exc2("存储过程的名字")
#示例2
14 # class exec: 15 # def __init__(self,proc_name): 16 # self.host='127.0.0.1' 17 # self.port=3306 18 # self.db='db1' 19 # self.charset='utf-8' 20 # self.proc_name=proc_name 21 # 22 # ex=exec('存储名称') 23 # ex2=exec('存储名称2') 24 # print(ex.__dict__) 25 # print(ex2.__dict__)
9.示例1, 现有如下代码， 会输出什么：
1 class People(object):
2 __name = "luffy"
3 __age = 18
4
5
6 p1 = People()
7 # print(p1.__name, p1.__age)
8 '''
9 AttributeError: 'People' object has no attribute '__name'
10 '''
11 # print(People.__dict__)
12 # print(p1._People__name,p1._People__age)
13 '''
14 {'__module__': '__main__', '_People__name': 'luffy', '_People__age': 18, '__dict__': <attribute '__dict__' of 'People' objects>, '__weakref__': <attribute '__weakref__' of 'People' objects>, '__doc__': None}
15 luffy 18
16 '''
10.示例2, 现有如下代码， 会输出什么：
1 class People(object):
2
3 def __init__(self,name):
4 print("__init__")
5
6 def __new__(cls, *args, **kwargs):
7 print("__new__")
8 print(args,kwargs)
9 return object.__new__(cls)
10
11 # People('alice')
12 '''
13 __new__
14 __init__
15 '''
11.请简单解释Python中 staticmethod（静态方法）和 classmethod（类方法）, 并分别补充代码执行下列方法。
静态方法：非绑定方法，类和对象都可调用
类方法：绑定给类的方法，类调用
1 class A(object):
2 def __init__(self,name):
3 self.name=name
4
5 def foo(self, x):
6 print("executing foo(%s, %s)" % (self,x))
7
8 @classmethod
9 def class_foo(cls, x):
10 print("executing class_foo(%s, %s)" % (cls,x))
11
12 @staticmethod
13 def static_foo(x):
14 print("executing static_foo(%s)" % (x))
15
16 # a = A('alice')
17 # a.foo('alice')
18 # A.class_foo('alice')
19 # a.static_foo('alice')
20 # A.static_foo('alice')
21 '''
22 executing foo(<__main__.A object at 0x000002A5FED12AC8>, alice)
23 executing class_foo(<class '__main__.A'>, alice)
24 executing static_foo(alice)
25 executing static_foo(alice)
26 '''
12.请执行以下代码，解释错误原因，并修正错误。
错误原因：@property 可将函数属性转化为数据属性
#先引一个例子讲解，后面的才是题目
'''@property的实现比较复杂，我们先考察如何使用。把一个getter方法变成属性，只需要加上@property就可以了，此时，@property本身又创建了另一个装饰器@score.setter，负责把一个setter方法变成属性赋值，于是，我们就拥有一个可控的属性操作： 注意到这个神奇的@property，我们在对实例属性操作的时候，就知道该属性很可能不是直接暴露的，而是通过getter和setter方法来实现的。''' class Student(object): @property def score(self): return self._score @score.setter def score(self,value): if not isinstance(value,int): raise ValueError("score must be an integer!") if value < 0 or value >100: raise ValueError("score must between 0 ~ 100!") self._score = value s = Student() s.score = 90 print(s.score) # s.score = 102 print(Student.mro())
#结果
"""
90
[<class '__main__.Student'>, <class 'object'>]
"""
1 class Dog(object):
2
3 def __init__(self,name):
4 self.name = name
5
6 @property
7 def eat(self):
8 print(" %s is eating" %self.name)
9
10 d = Dog("ChenRonghua")
11 # d.eat()
12 # d.eat
13.下面这段代码的输出结果将是什么？请解释。
1 class Parent(object): 2 x = 1 3 4 class Child1(Parent): 5 pass 6 7 class Child2(Parent): 8 pass 9 10 print(Parent.x, Child1.x, Child2.x) 11 Child1.x = 2 12 print(Parent.x, Child1.x, Child2.x) 13 Parent.x = 3 14 print(Parent.x, Child1.x, Child2.x) 15 16 # 1 1 1 继承自父类的类属性x，所以都一样，指向同一块内存地址 17 # 1 2 1 更改Child1，Child1的x指向了新的内存地址 18 # 3 2 3 更改Parent，Parent的x指向了新的内存地址
14.多重继承的执行顺序，请解答以下输出结果是什么？并解释。
super()表示的是 子类的mro()列表中的下一个
print(G.mro())
[<class '__main__.G'>, <class '__main__.D'>, <class '__main__.A'>, <class '__main__.B'>, <class 'object'>]
print(F.mro())
[<class '__main__.F'>, <class '__main__.C'>, <class '__main__.B'>, <class '__main__.D'>, <class '__main__.A'>, <class 'object'>]
1 class A(object):
2 def __init__(self):
3 print('A')
4 super(A, self).__init__()
5
6 class B(object):
7 def __init__(self):
8 print('B')
9 super(B, self).__init__()
10
11 class C(A):
12 def __init__(self):
13 print('C')
14 super(C, self).__init__()
15
16 class D(A):
17 def __init__(self):
18 print('D')
19 super(D, self).__init__()
20
21 class E(B, C):
22 def __init__(self):
23 print('E')
24 super(E, self).__init__()
25
26 class F(C, B, D):
27 def __init__(self):
28 print('F')
29 super(F, self).__init__()
30
31 class G(D, B):
32 def __init__(self):
33 print('G')
34 super(G, self).__init__()
35
36 # if __name__ == '__main__':
37 # g = G()
38 # f = F()
39 # print(G.mro())
40 # print(F.mro())
41
42 # G
43 # D
44 # A
45 # B
46 #
47 # F
48 # C
49 # B
50 # D
51 # A
52 # [<class '__main__.G'>, <class '__main__.D'>, <class '__main__.A'>, <class '__main__.B'>, <class 'object'>]
53 # [<class '__main__.F'>, <class '__main__.C'>, <class '__main__.B'>, <class '__main__.D'>, <class '__main__.A'>, <class 'object'>]
15.请编写一段符合多态特性的代码.
import abc
class Animal(metaclass =abc.ABCMeta): # metaclass= abc.ABCMeta #创建抽象类，只能被继承，不以被实例化
@abc.abstractmethod #@abc.abstractmethod 此方法要求继承类必须有这个函数
def eat(self):
pass
@abc.abstractmethod
def run(self):
pass
class People(Animal):
def __init__(self,name):
self.name = name
def eat(self):
print("%s is eating"% self.name)
def run(self):
print("%s is runing"% self.name)
peo = People("andy")
peo.eat()
peo.run()
16.很多同学都是学会了面向对象的语法，却依然写不出面向对象的程序，原因是什么呢？原因就是因为你还没掌握一门面向对象设计利器，
即领域建模，请解释下什么是领域建模，以及如何通过其设计面向对象的程序？
http://www.cnblogs.com/alex3714/articles/5188179.html 此blog最后面有详解
领域模型,顾名思义,就是需求所涉及的领域的一个建模,更通俗的讲法是业务模型。
定义：
需求到面向对象的桥梁
作用：
1.发掘重要的业务领域概念
2.建立业务领域概念之间的关系
方法：
从用例中找名词
领域建模的三字经方法:找名词、加属性、连关系。
参考：http://www.cnblogs.com/linhaifeng/articles/6182264.html#_label15
http://www.cnblogs.com/linhaifeng/articles/7341318.html
17.请写一个小游戏，人狗大站，2个角色，人和狗，游戏开始后，生成2个人，3条狗，互相混战，人被狗咬了会掉血，狗被人打了也掉血，
狗和人的攻击力，具备的功能都不一样。注意，请按题14领域建模的方式来设计类。
1 class Animal:
2 def __init__(self, name,life_value,aggressivity):
3 self.name = name
4 self.life_value = life_value
5 self.aggressivity = aggressivity
6
7 def attack(self,enemy):
8 enemy.life_value -= self.aggressivity
9
10
11 class People(Animal):
12 camp='home'
13 def attack(self,enemy):
14 super().attack(enemy)
15 print('from people')
16
17 class Dog(Animal):
18 camp='wo'
19 def attack(self,enemy):
20 super().attack(enemy)
21 print('from dog')
22
23 p1=People('alice',80,30)
24 p1=People('alex',80,30)
25 d1=Dog('w1',90,50)
26 d2=Dog('w2',90,50)
27 d3=Dog('w3',90,50)
28
29 # print(p1.life_value)
30 # d1.attack(p1)
31 # print(p1.life_value)
32
33 # print(d1.life_value)
34 # p1.attack(d1)
35 # print(d1.life_value)
18.编写程序, 在元类中控制把自定义类的数据属性都变成大写.
19.编写程序, 在元类中控制自定义的类无需init方法.
class Mymeta(type):
def __new__(cls, name,bases,class_dic):
update_class_dic = {}
for k,v in class_dic.items():
if not callable(v) and not k.startswith("__"):
update_class_dic[k.upper()] = v
else:
update_class_dic[k] = v
return type.__new__(cls,name,bases,update_class_dic)
def __call__(self, *args, **kwargs):
obj = object.__new__(self)
if args:
raise TypeError("must use keyword argument for key function")
for i in kwargs:
obj.__dict__[i]=kwargs[i]
return obj
class Chinese(metaclass = Mymeta):
country = "china"
tag = "legend of the dragon"
def talk(self):
print("%s is talking"%self.name)
print(Chinese.__dict__)
ch = Chinese(name = "alice",age = 18)
ch.talk()
print(ch.__dict__)
20.编写程序, 编写一个学生类, 要求有一个计数器的属性, 统计总共实例化了多少个学生.
1 class Student:
2 __count = 0
3 def __init__(self, name, age):
4 self.name = name
5 self.age = age
6 Student.__count += 1
7
8 @property
9 def talk(self):
10 print('%s is talking' % self.name)
11
12 @staticmethod
13 def tell_count():
14 print('总共实例化了 %s 人' % Student.__count)
15
16 # s1 = Student('alice', 18)
17 # s2 = Student('alex', 20)
18 # s3 = Student('egon', 28)
19 # Student.tell_count()
20 # s1.tell_count()
21 # s1.talk
22 # s2.talk
21.编写程序, A 继承了 B, 俩个类都实现了 handle 方法, 在 A 中的 handle 方法中调用 B 的 handle 方法
1 class B:
2 def handle(self):
3 print('from B handle')
4
5 class A(B):
6 def handle(self):
7 super().handle()
8 # print('from A handle')
9 # a=A()
10 # a.handle()
22.编写程序, 如下有三点要求：
1.自定义用户信息数据结构， 写入文件, 然后读取出内容, 利用json模块进行数据的序列化和反序列化
e.g
{
"egon":{"password":"123",'status':False,'timeout':0},
"alex":{"password":"456",'status':False,'timeout':0},
}
2.定义用户类，定义方法db，例如 执行obj.db可以拿到用户数据结构
3.在该类中实现登录、退出方法, 登录成功将状态(status)修改为True, 退出将状态修改为False(退出要判断是否处于登录状态).
密码输入错误三次将设置锁定时间(下次登录如果和当前时间比较大于10秒即不允许登录)
1 import json
2 import time
3 class User:
4 def __init__(self, name, password):
5 self.name = name
6 self.password = password
7 self.status = False
8 self.timeout = 0
9
10 @property
11 def db(self):
12 with open(self.name+'.txt', 'r', encoding='utf-8') as f:
13 data = json.load(f)
14 return data
15
16 def save(self):
17 obj={}
18 obj[self.name]={'password':self.password,'status':self.status,'timeout':self.timeout}
19 with open(self.name+'.txt', 'w', encoding='utf-8') as f:
20 json.dump(obj,f)
21
22 def login(self):
23 with open(self.name+'.txt', 'r+', encoding='utf-8') as f:
24 data = json.load(f)
25 count = 0
26 while count < 3:
27 password = input('password>>:').strip()
28 if password != data[self.name]['password']:
29 count += 1
30 continue
31 else:
32 if data[self.name]['timeout'] != 0:
33 if time.time() - data[self.name]['timeout'] > 10:
34 print('不允许登录了！超时')
35 break
36 else:
37 data[self.name]['status'] = True
38 f.seek(0)
39 f.truncate()
40 json.dump(data, f)
41 print('----welcome----')
42 break
43 else:
44 data[self.name]['status'] = True
45 f.seek(0)
46 f.truncate()
47 json.dump(data, f)
48 print('----welcome----')
49 break
50
51 else:
52 data[self.name]['timeout']=time.time()
53 f.seek(0)
54 f.truncate()
55 json.dump(data,f)
56
57
58 def quit(self):
59 with open(self.name+'.txt', 'r+', encoding='utf-8') as f:
60 data = json.load(f)
61 if data[self.name]['status'] == True:
62 data[self.name]['status'] = False
63 f.seek(0)
64 f.truncate()
65 json.dump(data, f)
66 else:
67 print('您是退出状态！')
68
69
70 # alex=User('alex','123')
71 # egon=User('egon','456')
72 # # alex.save()
73 # # egon.save()
74 # # print(alex.db)
75 # # print(egon.db)
76 # # alex.login()
77 # alex.quit()
78 # # egon.quit()
79 # print(alex.db)
80 # print(egon.db)
81
82 # alex.login()
83 # egon.login()
'''
23.用面向对象的形式编写一个老师角色, 并实现以下功能, 获取老师列表, 创建老师、删除老师、
创建成功之后通过 pickle 序列化保存到文件里，并在下一次重启程序时能读取到创建的老师, 例如程序目录结构如下.
'''
'''
.
|-- bin/
| |-- main.py 程序运行主体程序(可进行菜单选择等)
|-- config/
| |-- settings.py 程序配置(例如: 配置存储创建老师的路径相关等)
|-- db 数据存储(持久化, 使得每次再重启程序时, 相关数据对应保留)
| |-- teachers/ 存储所有老师的文件
| |-- ... ...
|-- src/ 程序主体模块存放
| |-- __init__.py
| |-- teacher.py 例如: 实现老师相关功能的文件
| |-- group.py 例如: 实现班级相关的功能的文件
|-- manage.py 程序启动文件
|-- README.md 程序说明文件
'''
'''
24.根据23 题, 再编写一个班级类, 实现以下功能, 创建班级, 删除班级, 获取班级列表、创建成功之后通过 pickle 序列化保存到文件里，
并在下一次重启程序时能读取到创建的班级.
25.根据 23题, 编写课程类, 实现以下功能, 创建课程(创建要求如上), 删除课程, 获取课程列表
26.根据23 题, 编写学校类, 实现以下功能, 创建学校, 删除学校, 获取学校列表
27.通过23题, 它们雷同的功能, 是否可以通过继承的方式进行一些优化
伪代码
class Behavior(object):
def fetch(self, keyword):
通过 keyword 参数 查询出对应的数据列表
class School(Behavior):
pass
class Teacher(Behavior):
pass
s = School()
t = Teacher()
s.fetch("school")
t.fetch("teacher")
'''
# ---------见作业----------
#网络编程练习题
1，什么是C/S架构？
C指的是client（客户端软件），S指的是Server（服务端软件）
一个C/S架构就是，实现服务端软件与客户端软件基于网络通信。
互联网中处处是C/S架构
如12306网站是服务端，你的浏览器是客户端（B/S架构也是C/S架构的一种）
腾讯作为服务端为你提供视频，你得下个腾讯视频客户端才能看它的视频）
C/S架构与socket的关系：
我们学习socket就是为了完成C/S架构的开发
2，互联网协议是什么？分别介绍五层协议中每一层的功能？
英语成为世界上所有人通信的统一标准，计算机之间的通信也应该有一个像英语一样的通信标准， 这个标准称之为互联网协议， 可以很明确地说：互联网协议就是计算机界的英语，网络就是物理链接 介质+互联网协议。 我们需要做的是，让全世界的计算机都学会互联网协议，这样任意一台计算机在 发消息时都严格按照协议规定的格式去组织数据，接收方就可以按照相同的协议解析出结果了，这就 实现了全世界的计算机都能无障碍通信。 按照功能不同，人们将互联网协议分为osi七层或tcp/ip五 层或tcp/ip四层（我们只需要掌握tcp/ip五层协议即可），这种分层就好比是学习英语的几个阶段， 每个阶段应该掌握专门的技能或者说完成特定的任务，比如：1、学音标 2、学单词 3、学语法 4、写作文
简单说，计算机之间的通信标准，就称为互联网协议
按照功能不同，人们将互联网协议分为osi七层或tcp/ip五层或tcp/ip四层
tcp/ip四层：应用层，传输层，网络层，网络接口层
tcp/ip五层：应用层，传输层，网络层，数据链路层，物理层
osi七层：应用层，表示层，会话层，传输层，网络层，数据链路层，物理层
3，基于tcp协议通信，为何建立链接需要三次握手，而断开链接却需要四次挥手
为什么要三次挥手？ 在只有两次“握手”的情形下，假设Client想跟Server建立连接，但是却因为中途连接请求的 数据报丢失了，故Client端不得不重新发送一遍；这个时候Server端仅收到一个连接请求，因此 可以正常的建立连接。但是，有时候Client端重新发送请求不是因为数据报丢失了，而是有可能 数据传输过程因为网络并发量很大在某结点被阻塞了，这种情形下Server端将先后收到2次请求， 并持续等待两个Client请求向他发送数据...问题就在这里，Cient端实际上只有一次请求，而 Server端却有2个响应，极端的情况可能由于Client端多次重新发送请求数据而导致Server端 最后建立了N多个响应在等待，因而造成极大的资源浪费！所以，“三次握手”很有必要！ 为什么要四次挥手？ 试想一下，假如现在你是客户端你想断开跟Server的所有连接该怎么做？ 第一步，你自己先停止向Server端发送数据，并等待Server的回复。但事情还没有完， 虽然你自身不往Server发送数据了，但是因为你们之前已经建立好平等的连接了， 所以此时他也有主动权向你发送数据；故Server端还得终止主动向你发送数据， 并等待你的确认。其实，说白了就是保证双方的一个合约的完整执行！
三次握手：client发送请求建立通道；server收到请求并同意，同时也发送请求建通道；client收到请求并同意，建立完成
四次挥手：client发送请求断开通道；server收到请求并同意，同时还回复client上一条消息；server也发送请求断开通道；client受到消息结束
为什么TCP协议终止链接要四次？ 1、当主机A确认发送完数据且知道B已经接受完了，想要关闭发送数据口（当然确认信号还是可以发）， 就会发FIN给主机B。 2、主机B收到A发送的FIN，表示收到了，就会发送ACK回复。 3、但这是B可能还在发送数据，没有想要关闭数据口的意思，所以FIN与ACK不是同时发送的， 而是等到B数据发送完了，才会发送FIN给主机A。 4、A收到B发来的FIN，知道B的数据也发送完了，回复ACK， A等待2MSL以后，没有收到B传来的 任何消息，知道B已经收到自己的ACK了，A就关闭链接，B也关闭链接了。
4，为何基于tcp协议的通信比基于udp协议的通信更可靠？
tcp:可靠 对方给了确认收到信息，才发下一个，如果没收到确认信息就重发
udp:不可靠 一直发数据，不需要对方回应
（1）UDP：user datagram protocol（用户数据报协议） 特点： ——1：面向无连接：传输数据之前源端和目的端不需要建立连接 ——2：每个数据报的大小都限制在64k（8个字节）以内 ——3：面向报文的不可靠协议（即发出去的数据不一定会接收到） ——4：传输速率快，效率高 ——5：现实生活实例：邮局寄件，实时在线聊天，视频协议等等 （2）TCP：transmission control protocol（传输控制协议） 特点： ——1：面向连接：传输数据之前需要建立连接 ——2：在连接过程中进行大量的数据传输 ——3：通过“三次握手”的方式完成连接，是安全可靠的协议 ——4：传输效率低，速度慢
?5，流式协议指的是什么协议，数据报协议指的是什么协议？
TCP协议，可靠传输
数据报协议: UDP协议，不可传输
也就是TCP和UDP的区别：
TCP是面向连接的，可靠的字节流服务
UDP是面向无连接的数据报服务
6，什么是socket？简述基于tcp协议的套接字通信流程流式协议：
Socket是应用层与TCP/IP协议族通信的中间软件抽象层，它是一组接口。在设计模式中， Socket其实就是一个门面模式，它把复杂的TCP/IP协议族隐藏在Socket接口后面， 对用户来说，一组简单的接口就是全部。 服务端：创建socket对象，绑定ip端口bind(), 设置最大链接数listen(), accept()与客户端 的connect()创建双向管道， send(), recv(),close() 客户端：创建socket对象，connect()与服务端accept()创建双向管道 , send(), recv(),close()
7，什么是粘包？ socket 中造成粘包的原因是什么？ 哪些情况会发生粘包现象？
粘包：数据粘在一起，主要因为：接收方不知道消息之间的界限，不知道一次性提取 多少字节的数据造成的数据量比较小，时间间隔比较短，就合并成了一个包， 这是底层的一个优化算法（Nagle算法）
8，基于socket开发一个聊天程序，实现两端互相发送和接收消息
服务端：
# _*_ coding: utf-8 _*_
# 8，基于socket开发一个聊天程序，实现两端互相发送和接收消息
import socket
ip_port = (‘127.0.0.1‘,8088)
link = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
link.bind(ip_port)
link.listen(5)
print("等待数据连接：。。》》")
# 阻塞直到有连接为止，有了一个新连接进来后，就会为这个请求生成一个连接对象
conn, addr = link.accept()
client_data = conn.recv(1024)
print("这是收到的消息：",client_data.decode(‘utf-8‘))
conn.send(client_data.upper())
conn.close()
link.close()
客户端
# _*_ coding: utf-8 _*_
# 8，基于socket开发一个聊天程序，实现两端互相发送和接收消息/
import socket
ip_port = (‘127.0.0.1‘,8088)
link = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
link.connect(ip_port)
print("开始发送数据")
cmd = input("请输入：>>").strip()
link.send(cmd.encode(‘utf-8‘))
data = link.recv(1028)
print(data)
link.close()
9，基于tcp socket，开发简单的远程命令执行程序，允许用户执行命令，并返回结果
服务端：
# _*_ coding: utf-8 _*_
# 9，基于tcp socket，开发简单的远程命令执行程序，允许用户执行命令，并返回结果
import socket
import struct
import subprocess
ip_port = (‘127.0.0.1‘,9999)
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.bind(ip_port)
sk.listen(5)
while True: #连接循环
conn,addr = sk.accept()
print(conn,addr)
while True: #通信循环
client_data = conn.recv(1024)
#处理过程
res = subprocess.Popen(client_data.decode(‘utf-8‘),shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout = res.stdout.read()
stderr = res.stderr.read()
# 先发报头(转成固定长度的bytes类型，那么怎么转呢？就用到了struct模块)
length = len(stdout)+len(stderr)
header = struct.pack(‘i‘,length)
conn.send(header)
conn.send(stderr)
conn.send(stdout)
conn.close()
sk.close()
conn.close()
sk.close()
客户端
# _*_ coding: utf-8 _*_
# 9，基于tcp socket，开发简单的远程命令执行程序，允许用户执行命令，并返回结果
import socket
import struct
ip_port = (‘127.0.0.1‘,9999)
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.connect(ip_port)
while True:
cmd = input("请输入命令:>>").strip()
if not cmd:continue
sk.send(cmd.encode(‘utf-8‘))
header_struct = sk.recv(1024)
unpack_res = struct.unpack(‘i‘,header_struct)
total_size = unpack_res[0]
recv_size = 0
total_data = b‘‘
while recv_size < total_size:
recv_data = sk.recv(1024)
recv_size += len(recv_data)
total_data += recv_data
print("返回的消息：%s"%total_data.decode(‘gbk‘))
sk.close()
10，基于tcp协议编写简单FTP程序，实现上传、下载文件功能，并解决粘包问题
客户端
# _*_ coding: utf-8 _*_
import socket
import struct
import json
downlaod_dir = r‘D:文件传输clientdownload‘
ip_port = (‘127.0.0.1‘,8808)
phone = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
phone.connect(ip_port)
while True:
cmd = input(">>>").strip() #get D:文件传输servera.avi
if not cmd:continue
phone.send(cmd.encode(‘utf-8‘))
#接受文件的内容，以写的方式打开一个新文件，接受服务端发来的文件内容，并写入客户端的新文件
#第一步，先收报头的长度，然后解包
obj = phone.recv(1024)
header_size = struct.unpack(‘i‘,obj)[0]
#第二部 再收报头
header_bytes = phone.recv(header_size)
#第三部，从报头中解析除对真实数据的描述信息
header_json = header_bytes.decode(‘utf-8‘)
header_dic = json.loads(header_json)
‘‘‘
header_dic = {
‘filename‘:filename, #a.avi
‘md5‘:‘dsdsd‘,
‘file_size‘:os.path.getsize(filename)
}‘‘‘
print(header_dic)
total_size = header_dic[‘file_size‘]
filename = header_dic[‘filename‘]
#第四步，接受真实的数据
with open(‘%s/%s‘%(downlaod_dir,filename),‘wb‘) as f:
recv_size = 0
# recv_data = b‘‘
while recv_size <total_size:
res = phone.recv(1024)
# recv_data += res
f.write(res)
recv_size += len(res)
print("总大小： %s
已经下载大小 ：%s"%(total_size,recv_size))
# print(recv_data.decode(‘utf-8‘))
phone.close()
服务端
# _*_ coding: utf-8 _*_
import subprocess
import socket
import struct
import json
import os
share_dir =r‘/文件传输/server/share‘
phone = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
ip_port = (‘127.0.0.1‘,8808)
phone.bind(ip_port)
phone.listen(5)
print("starting....")
while True: #链接循环
conn,client_addr = phone.accept()
print(client_addr)
while True: #通信循环
try:
#收命令
res = conn.recv(1024) #b‘get a.txt‘
if not res :continue
#解析命令，提取相应的参数
cmds = res.decode(‘utf-8‘).split()
filename = cmds[1]
#以读的方式打开文件，读取文件内容发送给客户端
# with open(filename,‘rb‘) as f:
# conn.s
#制定固定长度的报头
header_dic = {
‘filename‘:filename, #a.avi
‘md5‘:‘dsdsd‘,
‘file_size‘:os.path.getsize(r"%s/%s"%(share_dir,filename))
}
header_json = json.dumps(header_dic)
header_bytes = header_json.encode(‘utf-8‘)
#先发送报头的长度
conn.send(struct.pack(‘i‘,len(header_bytes)))
#再发报头
conn.send(header_bytes)
#再发真实的数据
with open(‘%s/%s‘%(share_dir,filename),‘rb‘) as f:
# conn.send(f.read())
for line in f:
conn.send(line)
except ConnectionResetError:
break
conn.close()
phone.close()
函数版本服务端
# _*_ coding: utf-8 _*_
import socket
import os
import struct
import pickle
dirname = os.path.dirname(os.path.abspath(__file__))
filepath = os.path.join(dirname, ‘share‘)
def get(cmds,conn):
filename = cmds[1]
file_path = os.path.join(filepath, filename)
if os.path.isfile(file_path):
header = {
‘filename‘: filename,
‘md5‘: ‘xxxxxx‘,
‘file_size‘: os.path.getsize(file_path)
}
header_bytes = pickle.dumps(header)
conn.send(struct.pack(‘i‘, len(header_bytes)))
conn.send(header_bytes)
with open(file_path, ‘rb‘) as f:
for line in f:
conn.send(line)
else:
conn.send(struct.pack(‘i‘, 0))
def put(cmds,conn):
pass
def run():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind((‘127.0.0.1‘, 8080))
server.listen(5)
print(‘starting...‘)
while True:
conn, client_addr = server.accept()
print(client_addr)
while True:
try:
res = conn.recv(1024)
if not res: continue
cmds = res.decode(‘utf-8‘).split()
if cmds[0] == ‘get‘:
get(cmds,conn)
elif cmds[0] == ‘put‘:
put(cmds,conn)
except ConnectionResetError:
break
conn.close()
server.close()
if __name__ == ‘__main__‘:
run()
函数版本客户端
# _*_ coding: utf-8 _*_
import socket
import struct
import pickle
import os
dirname = os.path.dirname(os.path.abspath(__file__))
filepath = os.path.join(dirname,‘download‘)
def get(client):
obj = client.recv(4)
header_size = struct.unpack(‘i‘, obj)[0]
if header_size == 0:
print(‘文件不存在‘)
else:
header_types = client.recv(header_size)
header_dic = pickle.loads(header_types)
print(header_dic)
file_size = header_dic[‘file_size‘]
filename = header_dic[‘filename‘]
with open(‘%s/%s‘ % (filepath, filename), ‘wb‘) as f:
recv_size = 0
while recv_size < file_size:
res = client.recv(1024)
f.write(res)
recv_size += len(res)
print(‘总大小：%s 已下载：%s‘ % (file_size, recv_size))
def put():
pass
def run():
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect((‘127.0.0.1‘,8080))
while True:
msg = input(">>>:").strip() # get a.txt
if not msg:continue
client.send(msg.encode(‘utf-8‘))
cmds = msg.split()
if cmds[0] == ‘get‘:
get(client)
elif cmds[0] == ‘put‘:
put()
client.close()
if __name__ == ‘__main__‘:
run()
面向对象版本服务端
# _*_ coding: utf-8 _*_
import socket
import os
import struct
import pickle
class TCPServer:
address_family = socket.AF_INET
socket_type = socket.SOCK_STREAM
listen_count = 5
max_recv_bytes = 8192
coding = ‘utf-8‘
allow_reuse_address = False
# 下载的文件存放路径
down_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), ‘share‘)
# 上传的文件存放路径
upload_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), ‘upload‘)
def __init__(self,server_address,bind_and_listen=True):
self.server_address = server_address
self.socket = socket.socket(self.address_family,self.socket_type)
if bind_and_listen:
try:
self.server_bind()
self.server_listen()
except Exception:
self.server_close()
def server_bind(self):
if self.allow_reuse_address:
self.socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
self.socket.bind(self.server_address)
def server_listen(self):
self.socket.listen(self.listen_count)
def server_close(self):
self.socket.close()
def server_accept(self):
return self.socket.accept()
def conn_close(self,conn):
conn.close()
def run(self):
print(‘starting...‘)
while True:
self.conn,self.client_addr = self.server_accept()
print(self.client_addr)
while True:
try:
res = self.conn.recv(self.max_recv_bytes)
if not res:continue
cmds = res.decode(self.coding).split()
if hasattr(self,cmds[0]):
func = getattr(self,cmds[0])
func(cmds)
except Exception:
break
self.conn_close(self.conn)
def get(self,cmds):
""" 下载
1.找到下载的文件
2.发送 header_size
3.发送 header_bytes file_size
4.读文件 rb 发送 send(line)
5.若文件不存在，发送0 client提示：文件不存在
:param cmds: 下载的文件 eg:[‘get‘,‘a.txt‘]
:return:
"""
filename = cmds[1]
file_path = os.path.join(self.down_filepath, filename)
if os.path.isfile(file_path):
header = {
‘filename‘: filename,
‘md5‘: ‘xxxxxx‘,
‘file_size‘: os.path.getsize(file_path)
}
header_bytes = pickle.dumps(header)
self.conn.send(struct.pack(‘i‘, len(header_bytes)))
self.conn.send(header_bytes)
with open(file_path, ‘rb‘) as f:
for line in f:
self.conn.send(line)
else:
self.conn.send(struct.pack(‘i‘, 0))
def put(self,cmds):
""" 上传
1.接收4个bytes 得到文件的 header_size
2.根据 header_size 得到 header_bytes header_dic
3.根据 header_dic 得到 file_size
3.以写的形式 打开文件 f.write()
:param cmds: 下载的文件 eg:[‘put‘,‘a.txt‘]
:return:
"""
obj = self.conn.recv(4)
header_size = struct.unpack(‘i‘, obj)[0]
header_bytes = self.conn.recv(header_size)
header_dic = pickle.loads(header_bytes)
print(header_dic)
file_size = header_dic[‘file_size‘]
filename = header_dic[‘filename‘]
with open(‘%s/%s‘ % (self.upload_filepath, filename), ‘wb‘) as f:
recv_size = 0
while recv_size < file_size:
res = self.conn.recv(self.max_recv_bytes)
f.write(res)
recv_size += len(res)
tcp_server = TCPServer((‘127.0.0.1‘,8080))
tcp_server.run()
tcp_server.server_close()
服务端
面向对象版本客户端
# _*_ coding: utf-8 _*_
import socket
import struct
import pickle
import os
class FTPClient:
address_family = socket.AF_INET
socket_type = socket.SOCK_STREAM
# 下载的文件存放路径
down_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), ‘download‘)
# 上传的文件存放路径
upload_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), ‘share‘)
coding = ‘utf-8‘
max_recv_bytes = 8192
def __init__(self, server_address, connect=True):
self.server_address = server_address
self.socket = socket.socket(self.address_family, self.socket_type)
if connect:
try:
self.client_connect()
except Exception:
self.client_close()
def client_connect(self):
self.socket.connect(self.server_address)
def client_close(self):
self.socket.close()
def run(self):
while True:
# get a.txt 下载 put a.txt 上传
msg = input(">>>:").strip()
if not msg: continue
self.socket.send(msg.encode(self.coding))
cmds = msg.split()
if hasattr(self,cmds[0]):
func = getattr(self,cmds[0])
func(cmds)
def get(self, cmds):
""" 下载
1.得到 header_size
2.得到 header_types header_dic
3.得到 file_size file_name
4.以写的形式 打开文件
:param cmds: 下载的内容 eg: cmds = [‘get‘,‘a.txt‘]
:return:
"""
obj = self.socket.recv(4)
header_size = struct.unpack(‘i‘, obj)[0]
if header_size == 0:
print(‘文件不存在‘)
else:
header_types = self.socket.recv(header_size)
header_dic = pickle.loads(header_types)
print(header_dic)
file_size = header_dic[‘file_size‘]
filename = header_dic[‘filename‘]
with open(‘%s/%s‘ % (self.down_filepath, filename), ‘wb‘) as f:
recv_size = 0
while recv_size < file_size:
res = self.socket.recv(self.max_recv_bytes)
f.write(res)
recv_size += len(res)
print(‘总大小：%s 已下载：%s‘ % (file_size, recv_size))
else:
print(‘下载成功！‘)
def put(self, cmds):
""" 上传
1.查看上传的文件是否存在
2.上传文件 header_size
3.上传文件 header_bytes
4.以读的形式 打开文件 send(line)
:param cmds: 上传的内容 eg: cmds = [‘put‘,‘a.txt‘]
:return:
"""
filename = cmds[1]
file_path = os.path.join(self.upload_filepath, filename)
if os.path.isfile(file_path):
file_size = os.path.getsize(file_path)
header = {
‘filename‘: os.path.basename(filename),
‘md5‘: ‘xxxxxx‘,
‘file_size‘: file_size
}
header_bytes = pickle.dumps(header)
self.socket.send(struct.pack(‘i‘, len(header_bytes)))
self.socket.send(header_bytes)
with open(file_path, ‘rb‘) as f:
send_bytes = b‘‘
for line in f:
self.socket.send(line)
send_bytes += line
print(‘总大小：%s 已上传：%s‘ % (file_size, len(send_bytes)))
else:
print(‘上传成功！‘)
else:
print(‘文件不存在‘)
ftp_client = FTPClient((‘127.0.0.1‘,8080))
ftp_client.run()
ftp_client.client_close()
客户端
11，基于udp协议编写程序，实现功能
服务端：
# _*_ coding: utf-8 _*_
import socket
ip_port = (‘127.0.0.1‘,8808)
udp_server_client = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
udp_server_client.bind(ip_port)
while True:
conn,addr = udp_server_client.recvfrom(1024)
print(conn,addr)
udp_server_client.sendto(conn.upper(),addr)
客户端：
# _*_ coding: utf-8 _*_
import socket
ip_port = (‘127.0.0.1‘,8808)
udp_server_client = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
while True:
cmd = input(">>>>").strip()
if not cmd:
continue
udp_server_client.sendto(cmd.encode(‘utf-8‘),ip_port)
back_cmd,addr = udp_server_client.recvfrom(1024)
print(back_cmd.decode(‘utf-8‘))
UDP不会发生粘包现象，下面举例说明
客户端
# _*_ coding: utf-8 _*_ import socket ip_port = (‘127.0.0.1‘,8989) client = socket.socket(socket.AF_INET,socket.SOCK_DGRAM) client.sendto(‘hello‘.encode(‘utf-8‘),ip_port) client.sendto(‘james‘.encode(‘utf-8‘),ip_port) client.close()
服务端
# _*_ coding: utf-8 _*_
import socket
ip_port = (‘127.0.0.1‘,8989)
server = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
server.bind(ip_port)
res1 = server.recvfrom(5)
print("res1:",res1)
res2 = server.recvfrom(5)
print("res2:",res2)
server.close()
12，执行指定的命令，让客户端可以查看服务端的时间
13，执行指定的命令，让客户端可以与服务的的时间同步
服务端
import socket
import time
import subprocess
server = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
server.bind(("127.0.0.1", 8080)) # 绑定IP、端口
while True:
cmd,addr = server.recvfrom(1024)
if not addr:
break
print(cmd.decode(), addr)
if cmd.decode() == "time": # 返回服务端时间
time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) # 当前日期时间
server.sendto(time_now.encode("gbk"), addr) # 发送
else: # 解析命令
obj = subprocess.Popen(cmd.decode(), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout = obj.stdout.read()
stderr = obj.stderr.read()
server.sendto(stdout+stderr, addr)
server.close()
客户端
import socket
import datetime
import os
client = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
ip_port = ("127.0.0.1",8081)
while True:
cmd = input("请输入命令:").strip()
if not cmd:continue
client.sendto(cmd.encode("utf-8"),ip_port)
res,addr = client.recvfrom(8096)
print(res.decode('gbk'))
res = res.decode("gbk")
#用户判断是否要同步时间
choice = input("1、同步（输入1）2、不是时间跳过(输入2):").strip()
while True:
if choice == "1":
# 转换字符串为时间
date_time = datetime.datetime.strptime(res, "%Y-%m-%d %H:%M:%S")
# 同步服务器时间
os.system("date %d-%d-%d"%(date_time.year,date_time.month,date_time.day)) #设置日期
os.system("time %d:%d:%d"%(date_time.hour,date_time.minute,date_time.second)) #设置时间
print(date_time)
elif choice == "2":
break
client.close()
基于元类实现单例模式,比如数据库对象,实例化时参数都一样,就没必要重复产生对象,浪费内存
class Mysql:
__instance=None
def __init__(self,host='127.0.0.1',port='3306'):
self.host=host
self.port=port
@classmethod
def singleton(cls,*args,**kwargs):
if not cls.__instance:
cls.__instance=cls(*args,**kwargs)
return cls.__instance
obj1=Mysql()
obj2=Mysql()
print(obj1 is obj2) #False
obj3=Mysql.singleton()
obj4=Mysql.singleton()
print(obj3 is obj4) #True
#参考
class MyList(list):
def __init__(self,*args):
for i in args:
if isinstance(i,int):
raise TypeError
super().__init__(args)
@property
def show_mid(self):
half_length = len(self)/2
if isinstance(half_length,int):
return [self[half_length-1],self[half_length]]
if isinstance(half_length,float):
return self[int(half_length)]
def clear(self):
pass
s = MyList("neo","alex","egon","jack","alice")
print(s.show_mid)
单例类：
class Singleton:
__instance = None
def __init__(self):
pass
def __new__(cls, *args, **kwargs):
if not cls.__instance:
cls.__instance = object.__new__(cls)
return cls.__instance
a = Singleton()
b = Singleton()
print(a)
print(b)
class A:
# def __init__(self):
# pass
#
__instance = None
@classmethod
def singleton(cls):
if not cls.__instance:
cls.__instance = super().__new__(cls)
return cls.__instance
a1 = A.singleton()
a2 = A.singleton()
print(a1,a2)
class MyList(list):
def __init__(self,*args,auth=True):
self.auth = auth
for i in args:
if not isinstance(i,str): #检测传入的内容必须是字符串
raise TypeError("only string is valid")
super().__init__(args) #调用list父方法的__init__初始化传入的列表参数
@property #特性
def show_mid(self):
length = len(self)
half_length = length/2 # 相除的结果是float型
if length%2 == 0:
return [self[int(half_length)-1],self[int(half_length)]]
else:
return self[int(half_length)]
def clear(self):
if self.auth:
super().clear()
else:
print("no auth to clear list")
s = MyList("neo","alex","egon","jack",auth=False)
print(s.show_mid)
s.clear()
print(s)