• 操作类题目
o 1. Python 交换两个变量的值
o 2. 在读文件操作的时候会使用 read、readline 或者 readlines，简述它们各自的作用
o 3. json 序列化时，可以处理的数据类型有哪些？如何定制支持 datetime 类型？
o 4. json 序列化时，默认遇到中文会转换成 unicode，如果想要保留中文怎么办？
o 5. 有两个磁盘文件 A 和 B，各存放一行字母，要求把这两个文件中的信息合并(按字母顺序排列)，输出到一个新文件 C 中。
o 6. 如果当前的日期为 20190530，要求写一个函数输出 N 天后的日期，(比如 N 为 2，则输出 20190601)。
o 7. 写一个函数，接收整数参数 n，返回一个函数，函数的功能是把函数的参数和 n 相乘并把结果返回。
o 8. 下面代码会存在什么问题，如何改进？
def strappend(num)：str='first'for i in range(num)：str+=str(i)return str
o 9. 一行代码输出 1-100 之间的所有偶数。
o 10. with 语句的作用，写一段代码？
o 11. Python 字典和 json 字符串相互转化方法
o 12. 请写一个 Python 逻辑，计算一个文件中的大写字母数量
o 13. 请写一段 Python连接Mongo数据库，然后的查询代码。
o 14.说一说Redis的基本类型
o 15. 请写一段 Python连接Redis数据库的代码。
o 16. 请写一段 Python连接Mysql数据库的代码。
o 17.了解Redis的事务么
o 18.了解数据库的三范式么？
o 19.了解分布式锁么
o 20.用 Python 实现一个 Reids 的分布式锁的功能
o 21.写一段 Python 使用 mongo 数据库创建索引的代码
1.Python 交换两个变量的值
答:
在 Python 中交换两个对象的值通过下面的方式即可
a , b = b ,a
但是需要强调的是这并不是元组解包，通过 dis 模块可以发现，这是交换操作的字节码是 ROT_TWO，意思是在栈的顶端做两个值的互换操作。
2.在读文件操作的时候会使用 read、readline 或者 readlines，简述它们各自的作用
答：
1).read() 每次读取整个文件，它通常用于将文件内容放到一个字符串变量中。
2)如果希望一行一行的输出那么就可以使用 readline()，该方法会把文件的内容加载到内存
3)对于对于大文件的读取操作来说非常的消耗内存资源，此时就可以通过 readlines 方法，将文件的句柄生成一个生产器，然后去读就可以了。
3.json 序列化时，可以处理的数据类型有哪些？如何定制支持 datetime 类型？
答：
可以处理的数据类型是 str、int、list、tuple、dict、bool、None, 因为 datetime 类不支持 json 序列化，所以我们对它进行拓展。
# 自定义时间序列化
import json
from datetime import datetime, date# JSONEncoder 不知道怎么去把这个数据转换成 json 字符串的时候
# ，它就会去调 default()函数,所以都是重写这个函数来处理它本身不支持的数据类型，
# default()函数默#认是直接抛异常的。
class DateToJson(json.JSONEncoder)：def default(self, obj)：if isinstance(obj, datetime)：return obj.strftime('%Y-%m-%d %H：%M：%S')elif isinstance(obj, date)：return obj.strftime('%Y-%m-%d')else：return json.JSONEncoder.default(self, obj)d = {'name'： 'cxa', 'data'： www.sychzs.cn()}
print(json.dumps(d, cls=DateToJson))
4.json 序列化时，默认遇到中文会转换成 unicode，如果想要保留中文怎么办？
答：
可以通过 json.dumps 的 ensure_ascii 参数解决，代码示例如下：
import json
a=json.dumps({"name"："张三"},ensure_ascii=False)
print(a)
5.有两个磁盘文件 A 和 B，各存放一行字母，要求把这两个文件中的信息合并(按字母顺序排列)，输出到一个新文件 C 中。
答：
#文件 A.txt 内容为 ASDCF
#文件 B.txt 内容为 EFGGTG
with open("A.txt") as f1：f1_txt = f1.readline()
with open("B.txt") as f2：f2_txt = f2.readline()
f3_txt = f1_txt + f2_txtf3_list = sorted(f3_txt)with open("C.txt", "a+") as f：f.write("".join(f3_list))
输出的文件 C 的内容为 ACDEFFGGGST
6.如果当前的日期为 20190530，要求写一个函数输出 N 天后的日期，(比如 N 为 2，则输出 20190601)。
答：这个题目考察的是 datetime 里的 timedelta 方法的使用，参数可选、默认值都为 0：datetime.timedelta(days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0) 通过这个参数可以指定不同的日期类型进行加减操作，这里我们需要改的是 days，代码如下
import datetimedef datetime_operate(n： int)：now = www.sychzs.cn() # 获取当前时间_new_date = now + datetime.timedelta(days=n) # 获取指定天数后的新日期new_date = _new_date.strftime("%Y%m%d") # 转换为指定的输出格式return new_dateif __name__ == '__main__'：print(datetime_operate(4))
7.写一个函数，接收整数参数 n，返回一个函数，函数的功能是把函数的参数和 n 相乘并把结果返回。
答：这个题目考查了闭包的使用代码示例如下，返回函数之类型是函数对象。
def mul_operate(num):def g(val):return num * valreturn gm = mul_operate(8)
print(m(5))
8.下面代码会存在什么问题，如何改进？
def strappend(num)：str='first'for i in range(num)：str+=str(i)return str
答： 首先不应该使用 Python 的内置类似 str 作为变量名这里我把它改为了 s,另外在Python,str 是个不可变对象，每次迭代都会生成新的存储空间，num 越大，创建的 str 对象就会越多，内存消耗越大。使用 yield 改成生成器即可, 还有一点就是命名规范的位置，函数名改为_分割比较好，完整的代码如下:
def str_append(num):s = 'first'for i in range(num):s += str(i)yield sif __name__ == '__main__':for i in str_append(3):print(i)
# 方法1print([i for i in range(1, 101) if i & 0x1 == 0])# 方法2：测试发现方法二效率更高print(list(range(2, 101, 2)))
10.with 语句的作用，写一段代码？
答:
with 语句适用于对资源进行访问的场合，确保不管使用过程中是否发生异常都会执行必要的“清理”操作，释放资源，比如文件使用后自动关闭、线程中锁的自动获取和释放等。
一般访问文件资源时我们会这样处理：
f = open('c：\test.txt', 'r')
data = www.sychzs.cn()
f.close()
这样写没有错，但是容易犯两个毛病：
1.如果在读写时出现异常而忘了异常处理。
2.忘了关闭文件句柄
以下的加强版本的写法：
f = open('c：\test.txt', 'r')
try：data = www.sychzs.cn()
finally：f.close()
以上的写法就可以避免因读取文件时异常的发生而没有关闭问题的处理了。代码长了一些。
但使用 with 有更优雅的写法：
with open(r'c：\test.txt', 'r') as f：data = www.sychzs.cn()
#with 的实现class Test：def __enter__(self)：print('__enter__() is call!')return selfdef dosomething(self)：print('dosomethong!')def __exit__(self, exc_type, exc_value, traceback)：print('__exit__() is call!')print(f'type：{exc_type}')print(f'value：{exc_value}')print(f'trace：{traceback}')print('__exit()__ is call!')with Test() as sample：pass#当对象被实例化时，就会主动调用__enter__()方法，任务执行完成后就会调用__exit__()方法，
#另外，注意到，__exit__()方法是带有三个参数的(exc_type, exc_value, traceback),
#依据上面的官方说明：如果上下文运行时没有异常发生，那么三个参数都将置为 None,
#这里三个参数由于没有发生异常，的确是置为了 None, 与预期一致.# 修改后不出异常了
class Test：def __enter__(self)：print('__enter__() is call!')return selfdef dosomething(self)：x = 1/0print('dosomethong!')def __exit__(self, exc_type, exc_value, traceback)：print('__exit__() is call!')print(f'type：{exc_type}')print(f'value：{exc_value}')print(f'trace：{traceback}')print('__exit()__ is call!')return Truewith Test() as sample：
11.Python 字典和 json 字符串相互转化方法
答:
在 Python 中使用 dumps 方法 将 dict 对象转为 Json 对象，使用 loads 方法可以将 Json 对象转为 dict 对象。
dic = {'a': 123, 'b': "456", 'c': "liming"}
json_str = json.dumps(dic)
dic2 = json.loads(json_str)
print(dic2)
打印结果：
'{"a": 123, "b": "456", "c": "liming"}'
{'a': 123, 'b': '456', 'c': 'liming'}
我们再来看一个特殊的例子
import json
dic = {'a'： 123, 'b'： "456", 'c'： "liming"}
dic_str = json.loads(str(dic).replace("'", "\""))
print(dic_str)
首先 json.loads(jsonstr) 这里面的参数只能是 jsonstr 格式的字符串.
当我们使用 str 将字典 dic 转化为字符串以后，得到的结果为:"{‘a’: 123, ‘b’: ‘456’, ‘c’: ‘liming’}"。
如果直接使用 json.loads(str(dic)) 你会发现出现错误，原因就是，单引号的字符串不符合Json的标准格式所以再次使用了 replace("’", “”")。得到字典
其实这个例子主要目的是告诉大家 Json 的标准格式是不支持单引号型字符串的，否则会出现以下错误。
json.decoder.JSONDecodeError: Expecting property name enclosed in double quot
12.请写一个 Python 逻辑，计算一个文件中的大写字母数量
答：
with open('A.txt') as fs：count = 0for i in www.sychzs.cn()：if i.isupper()：count += 1
print(count)
13.请写一段 Python连接Mongo数据库，然后的查询代码。
答:
# -*- coding: utf-8 -*-
import pymongo
db_configs = {'type': 'mongo','host': '地址','port': '端口','user': 'spider_data','passwd': '密码','db_name': 'spider_data'
}class Mongo():def __init__(self, db=db_configs["db_name"], username=db_configs["user"],password=db_configs["passwd"]):self.client = pymongo.MongoClient(f'mongodb://{db_configs["host"]}:db_configs["port"]')self.username = usernameself.password = passwordif self.username and self.password:self.db1 = self.client[db].authenticate(self.username, self.password)self.db1 = self.client[db]def find_data(self):# 获取状态为0的数据data = self.db1.test.find({"status": 0})gen = (item for item in data)return genif __name__ == '__main__':m = Mongo()print(m.find_data())
14.说一说Redis的基本类型
答:
Redis 支持五种数据类型： string（字符串） 、 hash（哈希）、list（列表） 、 set（集合） 及 zset(sorted set： 有序集合)。
15.请写一段 Python连接Redis数据库的代码
from redis import StrictRedis, ConnectionPool
redis_url="redis://:xxxx@112.27.10.168:6379/15"
pool = ConnectionPool.from_url(redis_url, decode_responses=True)
r= StrictRedis(connection_pool=pool)
16.请写一段 Python连接Mysql数据库的代码。
conn = pymysql.connect(host='localhost',
port=3306, user='root',
passwd='1234', db='user', charset='utf8mb4')#声明mysql连接对象
cursor=conn.cursor(cursor=pymysql.cursors.DictCursor)#查询结果以字典的形式
cursor.execute(sql语句字符串)#执行sql语句
conn.close()#关闭链接
17.了解Redis的事务么
答: 简单理解，可以认为 redis 事务是一些列 redis 命令的集合，并且有如下两个特点： 1.事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。 2.事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。 一般来说，事务有四个性质称为ACID，分别是原子性，一致性，隔离性和持久性。 一个事务从开始到执行会经历以下三个阶段：
开始事务
命令入队
执行事务
代码示例:
import redis
import sys
def run(): try:conn=redis.StrictRedis('192.168.80.41')# Python中redis事务是通过pipeline的封装实现的pipe=conn.pipeline()pipe.sadd('s001','a')sys.exit()#在事务还没有提交前退出，所以事务不会被执行。pipe.sadd('s001','b')pipe.execute()passexcept Exception as err:print(err)pass
if __name__=="__main__":run()
18.了解数据库的三范式么？
答:
经过研究和对使用中问题的总结，对于设计数据库提出了一些规范，这些规范被称为范式 一般需要遵守下面3范式即可: 第一范式（1NF）：强调的是列的原子性，即列不能够再分成其他几列。 第二范式（2NF）：首先是 1NF，另外包含两部分内容，一是表必须有一个主键；二是没有包含在主键中的列必须完全依赖于主键，而不能只依赖于主键的一部分。 第三范式（3NF）：首先是 2NF，另外非主键列必须直接依赖于主键，不能存在传递依赖。即不能存在：非主键列 A 依赖于非主键列 B，非主键列 B 依赖于主键的情况。
19.了解分布式锁么
答:
分布式锁是控制分布式系统之间的同步访问共享资源的一种方式。 对于分布式锁的目标，我们必须首先明确三点：
任何一个时间点必须只能够有一个客户端拥有锁。
不能够有死锁，也就是最终客户端都能够获得锁，尽管可能会经历失败。
错误容忍性要好，只要有大部分的Redis实例存活，客户端就应该能够获得锁。 分布式锁的条件 互斥性：分布式锁需要保证在不同节点的不同线程的互斥 可重入性：同一个节点上的同一个线程如果获取了锁之后，能够再次获取这个锁。 锁超时：支持超时释放锁，防止死锁 高效，高可用：加锁和解锁需要高效，同时也需要保证高可用防止分布式锁失效，可以增加降级。 支持阻塞和非阻塞：可以实现超时获取失败，tryLock(long timeOut) 支持公平锁和非公平锁
分布式锁的实现方案 1、数据库实现（乐观锁） 2、基于zookeeper的实现 3、基于Redis的实现（推荐）
20.用 Python 实现一个 Reids 的分布式锁的功能
答:
REDIS分布式锁实现的方式：SETNX + GETSET,NX是Not eXists的缩写，如SETNX命令就应该理解为：SET if Not eXists。 多个进程执行以下Redis命令：
SETNX www.sychzs.cn
如果 SETNX 返回1，说明该进程获得锁，SETNX将键 www.sychzs.cn 的值设置为锁的超时时间（当前时间 + 锁的有效时间）。 如果 SETNX 返回0，说明其他进程已经获得了锁，进程不能进入临界区。进程可以在一个循环中不断地尝试 SETNX 操作，以获得锁。
import time
import redis
from conf.config import REDIS_HOST, REDIS_PORT, REDIS_PASSWORDclass RedisLock:def __init__(self):self.conn = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, db=1)self._lock = 0self.lock_key = ""@staticmethoddef my_float(timestamp):"""Args:timestamp:Returns:float或者0如果取出的是None，说明原本锁并没人用，getset已经写入，返回0，可以继续操作。"""if timestamp:return float(timestamp)else:#防止取出的值为None，转换float报错return 0@staticmethoddef get_lock(cls, key, timeout=10):cls.lock_key = f"{key}_dynamic_lock"while cls._lock != 1:timestamp = time.time() + timeout + 1cls._lock = cls.conn.setnx(cls.lock_key, timestamp)# if 条件中，可能在运行到or之后被释放，也可能在and之后被释放# 将导致 get到一个None，float失败。if cls._lock == 1 or (time.time() > www.sychzs.cn_float(cls.conn.get(cls.lock_key)) andtime.time() > www.sychzs.cn_float(cls.conn.getset(cls.lock_key, timestamp))):breakelse:time.sleep(0.3)@staticmethoddef release(cls):if cls.conn.get(cls.lock_key) and time.time() < cls.conn.get(cls.lock_key):cls.conn.delete(cls.lock_key)def redis_lock_deco(cls):def _deco(func):def __deco(*args, **kwargs):cls.get_lock(cls, args[1])try:return func(*args, **kwargs)finally:cls.release(cls)return __decoreturn _deco@redis_lock_deco(RedisLock())
def my_func():print("myfunc() called.")time.sleep(20)if __name__ == "__main__":my_func()
21.写一段 Python 使用 mongo 数据库创建索引的代码:
答:
# -*- coding: utf-8 -*-
import pymongo
db_configs = {'type': 'mongo','host': '地址','port': '端口','user': 'spider_data','passwd': '密码','db_name': 'spider_data'
}class Mongo():def __init__(self, db=db_configs["db_name"], username=db_configs["user"],password=db_configs["passwd"]):self.client = pymongo.MongoClient(f'mongodb://{db_configs["host"]}:{db_configs["port"]}')self.username = usernameself.password = passwordif self.username and self.password:self.db1 = self.client[db].authenticate(self.username, self.password)self.db1 = self.client[db]def add_index(self):"""通过create_index添加索引"""self.db1.test.create_index([('name', pymongo.ASCENDING)], unique=True)def get_index(self,):"""查看索引列表"""indexlist=self.db1.test.list_indexes()for index in indexlist:print(index)if __name__ == '__main__':m = Mongo()m.add_index()print(m.get_index())
