AI学习-python

kiss strong

305人浏览 · 2026-06-21 10:29:18

kiss strong · 2026-06-21 10:29:18 发布

Python 的特点总结

优点

易学易用 语法简洁、可读性强，代码结构清晰，上手难度低。
开发效率高

内置能力：自带网络通信、文本处理、数据库、图形系统等丰富功能；
社区生态：拥有 Web 开发、科学计算、人工智能等各领域成熟第三方模块。

跨平台运行 同一份 Python 代码可在 Windows、Linux、macOS 等多操作系统直接运行。
应用领域广 ① 自动化脚本、桌面工具 ② 数据处理、科学计算（Pandas、Numpy） ③ 网络爬虫（Requests、BeautifulSoup） ④ Web 网站开发（Flask、Django） ⑤ AI / 机器学习（TensorFlow、PyTorch）

缺点

运行速度较慢 Python 属于解释型语言，运行时需要解释器实时翻译代码；而 C/C++ 会提前编译为机器码执行，因此 Python 运行效率更低。
代码难以加密 Python 为开源语言，发布的.py源码文件极易被逆向解析，对代码知识产权保护难度大。
内存消耗偏大 对比 C、C++ 这类紧凑型底层语言，Python 占用内存更高，不适合单片机、微控制器等硬件资源受限的设备。

语言分类

下载安装

卸载老版本

执行之前的安装程序，选择卸载，如果卸载出现如下情况，则使用管理员运行程序再次卸载即可

安装3.13.4版本

下载

Python Releases for Windows | Python.org

双击运行安装即可

函数参数

位置参数

就是正常的参数调用

def greet(name,age):
    print('Hello {},you are {} years old'.format(name,age))

greet('kitty',12)

关键字参数

def greet(name,age):
    print('Hello {},you are {} years old'.format(name,age))
greet(name='kitty',age=12)
greet(age=20,name='John')

注意

参数解包

使用*对参数进行解包，例如函数的入参是 a,b,c,现在有个列表[1,2,3]或元组（1,2,3），要调用这个函数则可以使用*对参数解包即可

def fun(a,b,c):
    print(a,b,c)

list_args = [11,22,334]
tuple_args = [22,2233,66]
fun(*list_args)
fun(*tuple_args)

修改函数入参为可变位置参数

def fun(*arg):
    print(arg)
    print('解包参数')
    print(*arg)

list_args = [11,22,334]
tuple_args = [22,2233,66]
fun(*list_args)
fun(*tuple_args)

数据容器_总结

有序与无序：

有序：列表(list)、元组(tuple)、字符串(str)—— 元素有顺序，可通过下标访问元素。
无序：集合(set)、字典(dict) —— 元素没有固定位置，不能用下标访问。

可修改：

可变：列表(list)、集合(set)、字典(dict) —— 可以对内容进行增、删、改操作。
不可变：元组(tuple) 、字符串(str) —— 内容固定，创建后无法修改。

可重复：

允许重复：列表(list) 、元组(tuple) 、字符串(str)
不允许重复：集合(set) 、字典(dict) 备注：字典的 key 是唯一的，但value 可重复

权限控制

三种访问权限

在 Python 中，我们可以给属性赋予三种权限，分别是：公有属性、受保护属性、私有属性

	写法	类内部访问	子类访问	外部访问
公有属性	属性名（name）	能	能	能
受保护属性	_属性名（_name）	能	能	能（不推荐）
私有属性	__属性名（__name）	能	不能	不能

class Person:
    def __init__(self, name, age, idcard):
        self.name = name # 公有属性：当前类内部、子类内部、类外部，可都可访问
        self._age = age # 受保护属性：当前类内部、子类内部，可以访问
        self.__idcard = idcard # 私有属性：仅能在当前类内部访问
    def speak(self):
    # 类的内部，可以访问任何权限的属性（公有属性、受保护属性、私有属性）。
        print(f"我叫：{self.name}，年龄：{self._age}，身份证：{self.__idcard}")
class Student(Person):
    def hello(self):
    # 子类的内部可以访问：公有属性、受保护属性
        print(f"我是学生，我叫：{self.name}，年龄：{self._age}")
    p1 = Person('张三', 18, '110101199001011234')
    print(p1.name)
    # 类的外部，仅能访问公有属性
    # 注意：如果在类的外部，强制访问【受保护的属性】，也能访问，但最好别这么做print(p1._age)
    # 强制访问【私有属性】会报错
    # print(p1.__idcard)
    # 扩展：Python 保护【私有属性】的方式，是重命名，例如：__idcard 属性，会被重命名为：_Person__idcard
    print(p1.__dict__)
    print(p1._Person__idcard)

getter 与 setter

在面向对象编程中，我们会把一些内部数据保护起来，但同时还想提供一个“安全的通道”让外部访问。这时候我们就用到：

getter：读取属性的方法
setter：修改属性的方法

在 Python 中：通过@property 和@xxx.setter 语法，把普通方法变成像属性一样使用的方法。

class Person:
    def __init__(self, name, age, idcard):
        self.name = name        # 公有属性：当前类内部、子类内部、类外部，可都可访问
        self._age = age         # 受保护属性：当前类内部、子类内部，可以访问
        self.__idcard = idcard  # 私有属性：仅能在当前类内部访问

    # 注册 age 属性的 getter 方法：当访问 Person 实例的 age 属性时，age方法会自动调用
    @property
    def age(self):
        return self._age

    # 注册 age 属性的 setter 方法：当给 Person 实例的 age 属性赋值时，age方法会自动调用
    @age.setter
    def age(self, value):
        if value <= 120:
            self._age = value
        else:
            print('年龄非法，已将年龄变为最大值120')
            self._age = 120

    # 注册 idcard 属性的 getter 方法：当访问 Person 实例的 idcard 属性时，会自动调用此方法
    @property
    def idcard(self):
        return self.__idcard[:6] + '********' + self.__idcard[-4:]

    # 注册 idcard 属性的 setter 方法，当 idcard 被修改时调用，内部会禁止修改身份证号并给出提示。
    @idcard.setter
    def idcard(self, value):
        print("抱歉，身份证号不允许随意修改，如有特殊需求，请联系管理员！")

p1 = Person('张三', 18, '110101199001011234')
print(p1.age)
print(p1.idcard)

# 测试修改age
p1._age = 19
print(p1.age)

# 测试修改idcard
p1.idcard = 'asd'

魔法方法

概念：以 __xxx__ 命名的特殊方法（双下划线开头和结尾）
特点：不需要手动调，在特定场景由 Python 自动调用

几个常用的魔法方法：

方法	调用时机
1️⃣__str__	当调用 print(对象)或str(对象) 时
2️⃣__len__	当调用len(对象)时
3️⃣__lt__	当执行对象1 < 对象2时
4️⃣__gt__	当执行对象1 > 对象2时
5️⃣__eq__	当执行对象1 == 对象2时
6️⃣__getattr__	当访问不存在的属性时

# 类中以双下划线开头和结尾的方法，叫魔法方法（魔术方法）。
# 魔法方法不需要手动调用，Python会在特定场景自动调用。
class Person:
    def __init__(self, name, age, gender):
        self.name = name
        self.age = age
        self.gender = gender

    # __str__ 方法，执行时机：当调用 print(对象)或str(对象) 时
    def __str__(self):
        return f'姓名：{self.name}， 年龄：{self.age}， 性别：{self.gender}'

    # __len__ 方法，执行时机：当调用len(对象)时
    def __len__(self):
        return len(self.__dict__)

    # __lt__方法，执行时机：当执行 对象1 < 对象2 时
    def __lt__(self, other):
        return self.age < other.age

    # __gt__方法，执行时机：当执行 对象1 > 对象2 时
    def __gt__(self, other):
        return self.age > other.age

    # __eq__方法，执行时机：当执行 对象1 == 对象2 时
    def __eq__(self, other):
        return self.__dict__ == other.__dict__

    # __getattr__方法，执行时机：当访问了对象不存在的属性时
    def __getattr__(self, item):
        print(f'您访问的{item}属性，不存在！')

# 创建Person实例
p1 = Person('张三', 18, '男')
p2 = Person('李四', 22, '女')

# 测试代码
print(p1)
print(str(p1))
print(len(p1))
print(p1 < p2)
print(p1 > p2)
print(p1 == p2)
print(p1.address)

object 类

object是所有类的最终祖先，所有的类，无论写与不写，都继承自object。
object提供了所有对象都会有的一组最基本方法

例如：

方法名	作用简述
__init__(self)	在对实例对象初始化时调用
__str__(self)	print(obj)或 int(obj)时调用
__class__	返回对象所属的类
......

备注：上述这些方法，如果我们不去重写，Python 会自动继承并使用默认版本。

# Python 中，所有的类都继承了 object 类，即：object 类是所有类的顶层父类。
class Person:
    def __init__(self, name, age, gender):
        self.name = name
        self.age = age
        self.gender = gender

# 验证一下：所有的类都继承了 object 类
print(issubclass(Person, object))
print(issubclass(int, object))
print(issubclass(str, object))
print(issubclass(list, object))
print(issubclass(tuple, object))
print(issubclass(bool, object))

# 因为 object 是所有类的父类，所以 Python 中的所有对象，都间接是 object 类的实例。
p1 = Person('张三', 18, '男')
print(isinstance(p1, object))
print(isinstance(100, object))
print(isinstance('hello', object))
print(isinstance(True, object))
print(isinstance(None, object))
print(isinstance([10, 20, 30], object))
print(isinstance({'吃饭','睡觉'}, object))

# 所有对象都继承了 object 类所提供的：各种属性和方法，从而保证每个对象都具备统一的基本能力。
for key in object.__dict__:
    print(key)

p1 = Person('张三', 18, '男')
print(p1.__dict__)  # 打印对象自己身上的东西
print(dir(p1))      # 打印对象能访问到的东西

print(p1.__str__())
print(p1)

多态

何为多态

多态就是“多种形态”，即：不同的对象调使用同一个方法名调用方法时，会表现出不同的行为。

标准多态

基于继承实现多态，又叫：『传统多态』或『标准多态』
常见于Java、C++ 这样的强类型语言中，Python支持『标准多态』

# 多态是指：同一个方法名，在不同的对象上调用时，能呈现出不同的行为。
# Python中支持：标准多态、鸭子多态

# 标准多态：
class Animal:
    def speak(self):
        print('动物正在发出声音')

class Dog(Animal):
    def speak(self):
        print('汪汪汪！')

class Cat(Animal):
    def speak(self):
        print('喵喵喵！')

# 注意Pig类没有继承Animal类
class Pig:
    def speak(self):
        print('哼哼哼！')

# make_sound函数要求：传入的对象，必须是 Animal 类型（或其子类型），才能保证可以调用到sepak方法
def make_sound(animal:Animal):
    animal.speak()

# 创建实例对象
a1 = Animal()
d1 = Dog()
c1 = Cat()

# 多态的体现：同一函数，不同对象 → 不同行为
make_sound(a1)  # 动物正在发出声音
make_sound(d1)  # 汪汪汪！
make_sound(c1)  # 喵喵喵！

# 按标准多态规则：Pig 没有继承 Animal，类型不匹配（会出现类型警告）
p1 = Pig()
make_sound(p1)  # 在其它语言中会报错，虽然 Python 中能运行，但这不属于标准多态

抽象类

概念：抽象类（Abstract Class）是一种不能被直接实例化的类，通常作为“规范”，让子类去继承并实现其中定义的抽象方法，本身只定义规范，不需要提供完整实现。
例如：动物会叫、飞行器会飞、支付方式会支付。

from abc import ABC, abstractmethod
#【抽象类】是一种不能直接实例化的类，它通常作为“规范”，让子类去继承，并实现其中定义的【抽象方法】。
# MustRun类一旦继承了ABC类，那MustRun类就是【抽象类】了
class MustRun(ABC):
    # run方法一旦被@abstractmethod装饰后，就变成了【抽象方法】
    @abstractmethod
    def run(self):
        pass

class Person(MustRun):
    def __init__(self, name, age, gender):
        self.name = name
        self.age = age
        self.gender = gender

    def run(self):
        print(f'我叫{self.name}，我在努力的奔跑！')

p1 = Person('张三', 18, '男')
p1.run()

内存分析

内存分为两个部分：栈内存、堆内存；变量在栈内存中，对象在堆内存中。

备注：强烈建议各位参考视频教程学习本小节

1️⃣Python 中变量里保存的不是存数据，而是指向堆中对象的引用（内存地址）。

2️⃣不可变对象：重新赋值会创建新对象

int 类的实例对象，是不可变对象，所以修改变量 a 时，会创建新对象，不会影响其他引用（b）

Python 中常见的不可变对象有：int 、float 、bool 、str 、tuple 、frozenset 、None。

Python 中常见的可变对象有：list 、dict 、set 、自定义类的实例对象。

3️⃣ 可变对象：修改内容不改变地址

4️⃣自定义类对象的内存表示

函数进阶

重新认识函数

函数也是对象

a1 = 100            # a1是int类的实例对象
a2 = 'hello'        # a2是str类的实例对象
a3 = [10, 20, 30]   # a3是list类的实例对象

# welcome函数是function类的实例对象
def welcome():
    print('你好啊')

print(type(a1))
print(type(a2))
print(type(a3))
print(type(welcome))

上述代码中welcome函数的内存示意图：

函数可以像对象一样，动态添加属性

def welcome():
    print("你好")

# 动态添加属性
welcome.desc = "这是一个用于打招呼的函数"
welcome.version = 1.0
print(welcome.__dict__)

# 调用函数
welcome()

函数可以赋值给变量

def welcome():
    print('你好啊！')

# 把函数对象赋值给变量
say_hello = welcome

# 通过变量调用函数
say_hello()

# 通过函数名调用函数
welcome()

上述代码的内存结构示意图：

可变参数 vs 不可变参数

def welcome(data):
    print(f'函数收到的data是：{data}，地址是：{id(data)}')
    data = 888
    print(f'被修改后的data是：{data}，地址是：{id(data)}')

a = 666
print(f'函数外侧a的值是：{a}，地址是：{id(a)}')
welcome(a)
print(f'函数调用后a的值是：{a}，地址是：{id(a)}')

def welcome(data):
    print(f'函数收到的data是：{data}，地址是：{id(data)}')
    data[2] = 99
    print(f'被修改后的data是：{data}，地址是：{id(data)}')

a = [10, 20, 30]
print(f'函数外侧a的值是：{a}，地址是：{id(a)}')
welcome(a)
print(f'函数调用后a的值是：{a}，地址是：{id(a)}')

函数也可以作为参数

def welcome():
    print('你好啊！')

def caller(f):
    print('caller函数开始调用')
    f()

caller(welcome)

函数也可以作为返回值

def welcome():
    print('你好啊')
    def show_msg(msg):
        print(msg)
    return show_msg

# result = welcome()
# result('尚硅谷')
welcome()('尚硅谷')

函数的多返回值

在return关键字后面写多个值，并且多个值之间用逗号隔开，Python 会自动把多个值打包成元组。

def calculate(x, y):
    res1 = x + y
    res2 = x - y
    return res1, res2 # 实际返回的是：(res1, res2)

result = calculate(10, 20)
r1, r2 = calculate(10, 20)

print(result)  # (30, -10)
print(r1)      # 30
print(r2)      # -10

参数的打包与解包

定义函数时，打包接收参数：

*形参名：打包所有的位置参数（会形成一个元组）。
**形参名：打包所有的关键字参数（会形成一个字典）。

调用函数时，解包传递参数：

*变量名：将元组拆解成一个一个独立的位置参数。
**变量名：将字典拆解一个一个key=value形式的关键字参数。

# 打包接收参数：
# *args    ：打包所有的位置参数（会形成一个元组）
# **kwargs ：打包所有的关键字参数（会形成一个字典）
def show_info(*args, **kwargs):
    print(args)
    print(kwargs)

nums = (10, 20, 30)
person = {'name': '张三', 'age': 18}

# 解包传递参数：
# *nums    ：将元组拆解成一个一个独立的位置参数
# **person ：将字典拆解一个一个 key=value 形式的关键字参数
show_info(*nums, **person)

高阶函数

当一个函数的『参数是函数』或者『返回值是函数』那该函数就是『高阶函数』

def welcome():
    print('你好啊！')
    
# caller函数的参数是函数，所以caller是高阶函数
def caller(f):
    print('call函数开始调用')
    f()
    
caller(welcome)


# outer函数的返回值是函数，所以outer函数是高阶函数
def outer():
    print('我是outer')
    def inner():
        print('我是inner')
    return inner

高阶函数的意义：

代码复用性高：可以把行为“独立出去”，传入不同函数实现不同逻辑。
能让函数更灵活，更通用。
高阶函数是：装饰器、闭包的基础。（后面会讲）

def info(msg):
    return '[提示]' + msg
def warn(msg):
    return '[警告]' + msg
def error(msg):
    return '[错误]' + msg

def log(fun, text):
    print(fun(text))

log(info, '文件保存成功！')
log(warn, '磁盘空间不足！')
log(error, '该用户不存在！')

条件表达式

表达式：执行后最终能得到一个值的代码，就是表达式，例如这些都是表达式：

3 + 5
'abc' * 3
5 > 3
'y' in 'Python'
len('hello')

表达式最终会形成一个值，可以写在任何一个需要值的地方。

条件表达式： 根据不同的条件，得到不同的值，又称：三元表达式，也叫：三目运算符。

1️⃣语法格式为：

结果1 if 条件 else 结果2

具体规则：如果条件为真，整个表达式的结果就是“结果1”，否则就是“结果2”

代码示例：

age = 20

# 传统if-else写法
if age >= 18:
    text = '成年'
else:
    text = '未成年'

# 条件表达式写法
text = '成年' if age >= 18 else '未成年·'
print(text)

什么时候适合用条件表达式？ —— 简单的二选一场景，可以让代码更紧凑

rain = True
food = '外卖' if rain else '出去吃'

is_vip = True
disscount = 0.8 if is_vip else 1.0

is_login = True
msg = '欢迎回来！' if is_login else '请先登录！'

匿名函数

概念：所谓『匿名函数』，就是没有名字的函数，它无需使用def关键字去定义。
语法：Python 中使用lambda关键字来定义『匿名函数』，格式为：lambda 参数: 表达式
使用场景： 当一个函数只用一次、只做一点点小事，使用匿名函数会更简洁。

# 使用普通函数实现计算效果
def add(x, y):
    return x + y

def sub(x, y):
    return x - y

def calculate(func, a, b):
    print(f'计算结果为：{func(a, b)}')

calculate(add, 30, 10)
calculate(sub, 30, 10)

# 使用匿名函数实现计算效果
def calculate(func, a, b):
    print(f'计算结果为：{func(a, b)}')

calculate(lambda x, y: x + y, 30, 10)
calculate(lambda x, y: x - y, 30, 10)

特点：

只能写一行，不能写多行代码。
不能写代码块（if、for、while）
冒号右边必须是表达式，且只能写一个表达式。
执行结果自动作为返回值。

is_adult = lambda age: '成年' if age >= 18 else '未成年'
print(is_adult(18))
print(is_adult(13))

几个数据处理函数（重要）

map 函数

map函数：对一组数据中的每一个元素，统一执行某种操作（加工），并生成一组新数据。
语法格式：map(操作函数, 可迭代对象)

# 统一数据处理
nums = [10, 20, 30, 40]
# map函数的返回值是一个迭代器对象，需要我们自己去手动遍历，或者手动类型转换
result = map(lambda x: x * 2, nums)
print(list(result))
print(nums)

# 字符串转换
names = ('python', 'java', 'js')
result = map(lambda x: x.upper(), names)
print(tuple(result))
print(names)

# 类型转换
str_number = {'1', '2', '3'}
result = map(int, str_number)
print(set(result))
print(str_number)

# 注意点：
# 1.延迟执行：map 不会立刻计算，只有在“需要结果”时才执行计算。
# 2.返回的是迭代器对象，且一旦遍历完成，就会被“耗尽”。
# 3.map不会影响元素数量。

nums = [10, 20, 30, 40]
result = list(map(lambda x: x * 2, nums))
print(result)
print(result)
print(result)
print(result)

filter 函数

filter函数：从一组数据中，筛选出符合条件的元素（过滤），并组成一组新数据。
语法格式：filter(过滤函数, 可迭代对象)

# 筛选数值
nums = [10, 20, 30, 40, 50]
result = filter(lambda n: n > 30, nums)
print(list(result))
print(nums)

# 筛选成年人
persons = [
    {'name': '张三', 'age': 15, 'gender': '男'},
    {'name': '李四', 'age': 16, 'gender': '女'},
    {'name': '王五', 'age': 17, 'gender': '男'},
    {'name': '李华', 'age': 18, 'gender': '女'},
    {'name': '赵六', 'age': 19, 'gender': '女'},
    {'name': '孙七', 'age': 20, 'gender': '男'}
]
result = filter(lambda p: p['age'] >= 18, persons)
print(list(result))

# 过滤一下非法字符串
names = ['张三', '', '李四', None, '王五']
result = filter(lambda n: n, names)
print(list(result))

# 注意点
# 1.延迟执行：filter不会立刻筛选，只有在“需要结果”时才执行。
# 2.返回的是迭代器对象，且一旦遍历完成就会被“耗尽”。
# 3.filter可能会影响元素数量。

# filter函数的特殊用法：如果不传递过滤函数，那么自动会过滤掉“假值”
data = [0, 1, '', 'hello', [], (), 5]
result = filter(None, data)
print(list(result))

sorted 函数

sorted函数：对一组数据进行排序，返回一组新数据。
语法格式：sorted(可迭代对象, key=xxx, reverse=xxx)

# 数字排序
nums = [30, 40, 20, 10]
result = sorted(nums, reverse=True)
print(result)

# 按照字符串的长度去排序
names = ['python', 'sql', 'java']
result = sorted(names, key=len, reverse=True)
print(result)

# 根据字典中的某个字段进行排序
persons = [
    {'name':'张三', 'age':15, 'gender':'男'},
    {'name':'李四', 'age':17, 'gender':'女'},
    {'name':'王五', 'age':19, 'gender':'男'},
    {'name':'李华', 'age':20, 'gender':'女'},
    {'name':'赵六', 'age':18, 'gender':'女'},
    {'name':'孙七', 'age':16, 'gender':'男'}
]
result = sorted(persons, key=lambda p: p['age'], reverse=True)
print(result)

# 我们之前讲的max函数、min函数，也可以传递key参数，用于设置筛选依据
result1 = max(persons, key=lambda p: p['age'])
result2 = min(persons, key=lambda p: p['age'])
print(result1)
print(result2)

reduce 函数

reduce函数：将一组数据不断“合并”，最终归并成一个结果。
语法格式：reduce(合并函数, 可迭代对象, 初始值)。

备注：reduce函数需要从functools模块中引入才能使用。

# 从 functools 模块中引入 reduce
from functools import reduce

# 数值统计
nums = [1, 2, 3, 4, 5]
result = reduce(lambda a, b: a + b, nums, 10)
print(result)

# 字符串拼接
str_list = ['ab', 'cd', 'ef']
result = reduce(lambda a, b: a + b, str_list)
print(result)

列表推导式

概念：用一条简洁语句，从可迭代对象中，生成新列表的语法结构。
语法格式：[ 表达式 for 变量 in 可迭代对象 ]

# 需求：让nums列表中所有的元素，都变为原来的2倍

# 方式一：用map函数
nums = [10, 20, 30, 40]
result = list(map(lambda n: n * 2, nums))
print(result)

# 方式二：使用for循环结合append方法
nums = [10, 20, 30, 40]
result = []
for n in nums:
    result.append(n * 2)
print(result)

# 方式三：使用列表推导式(列表推导式就是上面 for + append 的简写形式)
nums = [10, 20, 30, 40]
result = [n * 2 for n in nums]
print(result)

# 带条件的列表推导式
nums = [10, 20, 30, 40]
result = [n * 2 for n in nums if n > 20]
print(result)

# 字典推导式
names = ['张三', '李四', '王五']
scores = [60, 70, 80]
result = {names[i]: scores[i] for i in range(len(names))}
print(result)

# 集合推导式
names = ['张三', '李四', '王五']
result = {n + '！' for n in names}
print(result)

names = ['张三', '李四', '王五']
# 注意：Python中没有元组推导式，下面这种写法叫：生成器（后面会仔细讲）
result = (n + '！' for n in names)
print(result)

常用内置函数梳理

以下函数是截至目前，我们需要掌握的一些常用函数，注意：下面这些并不是所有，我们后面还会学习到更多的常用内置函数。（详细讲解请参考视频教程）

输入与输出

函数 / 参数	功能说明
print	输出指定内容
objects	要输出的内容
sep	输出多个内容时的分隔符
end	结尾追加的内容
file	输出位置（默认控制台）
flush	是否立即刷新输出
input()	获取用户输入

类型转换

函数	功能说明
int()	转为整数
float()	转为浮点数
str()	转为字符串
bool()	转为布尔值
list()	转为列表
tuple()	转为元组
set()	转为集合
dict()	转为字典

数学相关

函数	功能说明
abs()	取绝对值
round()	银行家舍入法：小于5舍，大于5入，等于5看奇偶（奇入偶舍）
pow(a, b)	计算 a 的 b 次方
pow(a, b, c)	计算 a 的 b 次方后，再对 c 取模
divmod(a, b)	返回 (商, 余数)
max()	最大值（支持 key）
min()	最小值（支持 key）
sum()	求和
map()	加工一组数据
filter()	按条件过滤数据
reduce()	合并计算（需 functools.reduce）
sorted()	排序（支持 key）

数据容器相关

函数	功能说明
len()	获取容器中元素个数
range()	生成数字序列（常用于循环）
enumerate()	为序列添加索引
zip()	将多个序列一一配对

类型判断与对象相关

函数	功能说明
type()	查看对象类型
isinstance(obj, type)	判断对象是否属于某个类型
issubclass(A, B)	判断类 A 是否为类 B 的子类
id()	查看对象的内存地址

逻辑判断相关

函数	功能说明
all()	所有元素为真时返回 True
any()	至少一个为真返回 True

字符串辅助相关

函数	功能说明
ord()	获取字符的 Unicode 编码值
chr()	将 Unicode 编码值转换为字符

浅拷贝 vs 深拷贝

为什么要拷贝？

赋值语句：b = a ，只是让b 指向和 a 一样的对象。
如果a和b指向的是一个可变对象，那通过b修改后，再通过a访问到的数据也是变化后的。

直接赋值

在如下代码中：nums2 = nums1 不是复制！是两个变量指向同一个列表，并且修改任何一个，都会影响另一个。

nums1 = [10, 20, 30, 40]
nums2 = nums1
nums2[3] = 99

print(nums1[3]) # 99
print(nums2[3]) # 99

浅拷贝

浅拷贝会创建一个新的外层容器，但内部的元素仍然引用原来的对象

import copy
nums1 = [10, 20, 30, 40]
nums2 = copy.copy(nums1)
nums2[3] = 99

print(nums1[3]) # 40
print(nums2[3]) # 99

浅拷贝存在的问题：嵌套数据仍然是共享的，修改嵌套数据会互相影响

import copy

nums1 = [10, 20, 30, [40, 50]]
nums2 = copy.copy(nums1)
nums2[3][0] = 99

print(nums1[3][0])
print(nums2[3][0])

深拷贝

创建一个新的外层容器，同时对内部所有【可变对象】进行递归复制（不可变对象不复制，继续引用）。

import copy

nums1 = [10, 20, 30, [40, 50]]
nums2 = copy.deepcopy(nums1)
nums2[3][0] = 99

print(nums1[3][0])
print(nums2[3][0])

特点：

深拷贝可以彻底消除数据之间的相互影响。
深拷贝遇到【不可变对象】不会复制，会直接引用。

注意点：

深拷贝只复制可变对象，不可变对象会直接引用。
元组中如果只包含不可变对象，则深拷贝没有效果。

import copy
a = 666
# a是不可变对象，即便调用deepcopy也不会深拷贝，会直接引用
b = copy.deepcopy(a)

print(id(a))
print(id(b))

import copy
nums1 = (10, 20, 30, [40, 50])
# nums1元组中只包含不可变对象，即便调用deepcopy也不会深拷贝
nums2 = copy.deepcopy(nums1)

print(id(nums1))
print(id(nums2))

异常

模版

class MyException(Exception):
    def __init__(self, value):
        self.value = value
    def __str__(self):
        return repr(self.value)

try:
    a = int(input())
    print("ok")
except (ValueError,Exception) as e :
    if isinstance(e,ValueError):
        raise MyException('自定义异常')
    else:
        raise e
except BaseException as e:
    print(f"error{e}")
else:
    print("程序正常ok1")
finally:
    print("ok2")

包（管理模块）与模块（管理功能）

模块概述：
1.在 Python 中，一个.py文件就是一个模块（Module）。
2.模块中可以包含：变量、函数、类、等很多内容。
3.通常把能够实现某一特定功能的代码，集中放在一个模块中（模块就类似于一个工具箱）。
4.模块可以提高代码的可维护性 与 可复用性，还可以避免命名冲突。

模块的分类：
Python 中的模块分为三类，分别是：标准库模块、自定义模块、第三方模块。

模块命名注意点：
 1.要符合标识符命名规则
 2.模块名（.py文件名）区分大小写
 3.不要与标准库模块同名（一旦同名，Python会优先引入标准库模块）

自定义模块

引入方式

import 模块名
import 模块名 as 别名
from 模块名 import 具体内容1, 具体内容2, ......
from 模块名 import 具体内容1 as 别名1, 具体内容2 as 别名2, ......
from 模块名 import *

使用方法

import 模块名

import order
import pay

print(order.max_order_amount)
order.create_order()
order.cancel_order()
order.show_info()
print('*' * 10)
print(pay.timeout)
pay.wechat_pay()
pay.ali_pay()
pay.show_info()

import 模块名 as 别名

import order as dd
import pay as zf

print(dd.max_order_amount)
dd.create_order()
dd.cancel_order()
dd.show_info()
print('*' * 10)
print(zf.timeout)
zf.wechat_pay()
zf.ali_pay()
zf.show_info()

from 模块名 import 具体内容1, 具体内容2, ......

from order import max_order_amount, show_info
from pay import wechat_pay, ali_pay

print(max_order_amount)
show_info()
wechat_pay()
ali_pay()

from 模块名 import 具体内容1 as 别名1, 具体内容2 as 别名2, ......

from order import max_order_amount as max_amt, show_info as show1
from pay import timeout as tm, show_info as show2
print(max_amt)
print(tm)
show1()
show2()

from 模块名 import *

from order import *
from pay import *

max_order_amount = 10

print(max_order_amount)
create_order()
cancel_order()
show_info()

print(timeout)
wechat_pay()
ali_pay()
show_info()

关于__all__：
● 在 Python 模块中，可通过 __all__ 来控制【from 模块 import *】能导入哪些内容。
● __all__ 的值可以是：列表、元组。

# 订单最大金额
max_order_amount = 1000000

# 创建订单
def create_order():
    print('订单创建成功！')

# 关闭订单
def cancel_order():
    print('订单关闭成功！')

# 提示函数
def show_info():
    print('我是来自【订单】模块的提示！')
# 外部通过 from order import * 只能引入这两个，但仅限于这种， from order import max_order_amount 这种还是可以引入别的
__all__ = ('cancel_order', 'show_info')

关于__name__:
● __name__是每个 Python 模块（.py文件）都拥有的一个内置变量。
● 它的具体值取决于模块的运行方式：
      1.作为主程序直接运行，__name__ 的值是 __main__
      2.作为模块被导入到其他程序中运行，__name__的模块的文件名（不带.py）。

# 支付超时时间
timeout = 1800

# 微信支付
def wechat_pay():
    print('我是微信支付！')

# 支付宝支付
def ali_pay():
    print('我是支付宝支付！')

# 提示函数
def show_info():
    print('我是来自【支付】模块的提示！')

# print('我是pay模块打印的内容', __name__)

# 测试代码 如果是在pay里面运行则__name__的值是__main__,如果是外部引入则__name__的值是模块名
if __name__ == '__main__':
    wechat_pay()
    ali_pay()

标准库模块

标准库模块：随着 Python 一起安装在我们电脑上的那些模块。

有一些标准库模块是用C语言实现的，这些用C语言实现的模块，又称：【内置模块】

import copy     # 拷贝对象（浅拷贝、深拷贝）
import os       # 操作系统相关操作（文件、文件夹、路径系统层面的操作）
import random   # 随机数相关（生成随机数、随机选择、洗牌）
# print(copy.__file__)
# print(os.__file__)
# print(random.__file__)

# 如下的这些模块，属于内置模块（无法看到具体实现，也没有__file__属性）
import time     # 时间相关操作（获取时间、延时、格式化时间等。）
import math     # 数学相关（开平方、取绝对值 等等）
import sys      # Python 解释器相关操作

# 创建一个文件夹
os.mkdir('demo')

# 随机选择一个人
names = ['张三', '李四', '王五', '李华']
print(random.choice(names))

# 洗牌
names = ['张三', '李四', '王五', '李华']
random.shuffle(names)
print(names)

# 休眠
time.sleep(2)
print('ok')

# 获取当前时间
print(time.strftime('%H:%M:%S'))
print(time.strftime('%p %I:%M:%S'))

# 开平方
print(math.sqrt(4))

# 取绝对值
print(math.fabs(-11.2))

# 获取Python解释器的版本
print(sys.version)

如果点进去查看方法是*开头的就是c语言写的，无法查询原代码

包

包概述：
1.在 Python 中，【包含 __init__.py 的文件夹】就是一个包（Package）。
2.我们通常会把【某个特定功能相关的所有模块】放入一个包中。
3.使用包可以进一步提升代码的：可维护性、可复用性，便于管理大型项目。

包与模块的关系：
1.一个模块就是一个.py文件 ，包是用来“管理模块”的目录(文件夹)。
2.一个包中可以有多个模块，也可以有多个子包。

包的分类：
Python 中的包分为三类，分别是：标准库包、自定义包、第三方包。

包命名注意点：
1.要符合标识符命名规范。
2.包名区分大小写（建议全部使用小写字母）
3.不要与标准库包同名。

引入方式

import 包名.模块名
import 包名.模块名 as 别名
from 包名.模块名 import 具体内容1, 具体内容2, ......
from 包名.模块名 import 具体内容1 as 别名1, 具体内容2 as 别名2, ......
from 包名.模块名 import *
from 包名 import 模块名
from 包名 import 模块名 as 别名
from 包名 import *

使用方法

import 包名.模块名

import trade.pay
trade.order.create_order()
trade.pay.wechat_pay()

import 包名.模块名 as 别名

import trade.order as dd
import trade.pay as zf

dd.create_order()
zf.wechat_pay()

from 包名.模块名 import 具体内容1, 具体内容2, ......

from trade.order import max_order_amount, create_order
from trade.pay import timeout, wechat_pay

print(max_order_amount)
print(timeout)
create_order()
wechat_pay()

from 包名.模块名 import 具体内容1 as 别名1, 具体内容2 as 别名2, ......

from trade.order import max_order_amount as max_amt, create_order
from trade.pay import timeout, wechat_pay as w_pay

print(max_amt)
print(timeout)
create_order()
w_pay()

from 包名.模块名 import *

from trade.order import *
from trade.pay import *

# print(max_order_amount)
create_order()
cancel_order()
show_info()

print(timeout)
wechat_pay()
ali_pay()
show_info()

from 包名 import 模块名

from trade import order, pay
order.create_order()
pay.wechat_pay()

from 包名 import 模块名 as 别名

from trade import order as dd, pay as p
dd.create_order()
p.wechat_pay()

from 包名 import *

from trade import *
print(order.max_order_amount)
order.create_order()
print(pay.timeout)
pay.wechat_pay()

import 包名

import trade
print(trade.a)
print(trade.b)
trade.order.create_order()
trade.pay.wechat_pay()

测试一下引入子包

from trade.hello.h1 import say_hello
say_hello()

演示一下标准库包的使用

from collections import Counter
names = ['张三', '李四', '王五', '李华', '张三', '李四', '张三', '王五']
result = Counter(names)
print(result)

第三方包与模块

类似于java的maven

官网仓库地址 PyPI · The Python Package Index

环境

全局环境&虚拟环境

镜像加速

清华大学 https://pypi.tuna.tsinghua.edu.cn/simple
阿里云 https://mirrors.aliyun.com/pypi/simple
中国科技大学 https://pypi.mirrors.ustc.edu.cn/simple

使用方法

pip install -i https://mirrors.aliyun.com/pypi/simple numpy

pip 常用命令表

命令	说明
`pip install 包名`	安装指定的包。
`pip install -i 镜像地址包名`	临时使用镜像地址安装指定包。
`pip config set global.index-url 地址`	设置 pip 所使用的镜像地址。
`pip config list`	查看当前环境的 pip 配置。
`pip config unset global.index-url`	让 pip 恢复使用官方默认的地址。
`pip list`	列出当前环境中，已安装的所有第三方包。
`pip uninstall 包名`	从当前环境中卸载指定的第三方包。

全局安装

全局项目创建

全局安装numpy

随便找个命令行安装即可

pip install numpy

在项目中使用

虚拟环境安装

创建项目

安装依赖

打开虚拟环境

或者

cmd命令

安装numpy

切换环境和安装依赖

切换环境

命令行退出虚拟环境只要输入deactivate即可

安装依赖

迭代器

知识点

能被 for 循环遍历的对象，被称为：可迭代对象(iterable)

names = ['张三', '李四', '王五']
citys = ('北京', '上海', '深圳')
msg = 'hello'
age = 10
def test():
    pass

for item in test:
    print(item)

可迭代对象(iterable) 能调用到 __iter__ 方法。

names = ['张三', '李四', '王五']
citys = ('北京', '上海', '深圳')
msg = 'hello'
age = 10
def test():
    pass

names.__iter__()
citys.__iter__()
msg.__iter__()

print(hasattr(names, '__iter__'))
print(hasattr(citys, '__iter__'))
print(hasattr(msg, '__iter__'))
print(hasattr(age, '__iter__'))
print(hasattr(test, '__iter__'))

调用 __iter__ 方法会得到：迭代器(iterator)
备注1：__iter__ 是一个魔法方法，当调用 iter 函数时，__iter__ 会自动调用。
备注2：可迭代对象.__iter__()  等价于  iter(可迭代对象)。
备注3：如果 iter(obj) 能得到一个迭代器(iterator)，那 obj 就是可迭代对象。

names = ['张三', '李四', '王五']
citys = ('北京', '上海', '深圳')
msg = 'hello'

print(names.__iter__())
print(citys.__iter__())
print(msg.__iter__())

print(iter(names))
print(iter(citys))
print(iter(msg))

迭代器(iterator)拥有 __next__ 方法，每次调用都会根据当前的状态，返回下一个元素。
备注1：迭代器.__next__()  等价于  next(迭代器)。
备注2：当所有元素全都取出后，若继续调用 __next__ 方法，Python会抛出 StopIteration 异常。

names = ['张三', '李四', '王五']
it = iter(names)
print(it.__next__())
print(it.__next__())
print(it.__next__())
print(it.__next__())

print(next(it))
print(next(it))
print(next(it))
print(next(it))

迭代器(iterator)也拥有 __iter__ 方法，并且其返回值是迭代器自身。
这么设计的原因如下：让 for 循环也能遍历迭代器（即：为了让 iter(迭代器) 不出错）。

names = ['张三', '李四', '王五']
it = iter(names)
print(it)

result = iter(it)
print(result)
x = iter(result)
print(x)
it = iter(names)
for item in it:
    print(item)

迭代器协议

  1.能被 iter() 接受
  2.能被 next() 一步一步取值

for循环背后的工作逻辑

names = ['张三', '李四', '王五']

# 编写for循环遍历names列表
for item in names:
    print(item)

# for循环背后的逻辑
#  调用【可迭代对象的__iter__方法】获取到一个迭代器(iterator)
it = iter(names)
# 开启一个无限循环
while True:
    try:
        # 调用__next__方法，获取下一个元素
        item = next(it)
        print(item)
    except StopIteration:
        # 捕获 StopIteration 异常，随后结束循环
        break

迭代器优势

1.迭代器是惰性计算，不会一次性生成所有结果，所以能显著降低内存占用。
2.当数据量很大，不确定要用多少结果时，推荐使用迭代器。

import tracemalloc

# 使用迭代器实现
class Fibo:
    def __init__(self, total):
        # 要生成多少个数
        self.total = total
        # 当前生成到第几个了（计数器，指针）
        self.index = 0
        # 初始的两个值
        self.pre = 1
        self.cur = 1

    def __iter__(self):
        return self

    def __next__(self):
        # 当生成足够数量后，抛出StopIteration异常
        if self.index >= self.total:
            raise StopIteration
        # 前两项都是1
        if self.index < 2:
            value = 1
        else:
            # 新的结果等于前两项的和
            value = self.pre + self.cur
            # 更新一下pre和cur
            self.pre = self.cur
            self.cur = value
        # 计数器+1
        self.index += 1
        # 返回value
        return value


# 不用迭代器实现
def fibo(total):
    if total <= 0:
        return []
    if total == 1:
        return [1]
    nums = [1, 1]
    for i in range(2, total):
        nums.append(nums[-1] + nums[-2])
    return  nums

# 看内存占用
tracemalloc.start()
f1 = Fibo(0)
m = tracemalloc.get_traced_memory()[1]
print(f'内存占用是：{m / 1024 / 1024}MB')

tracemalloc.start()
f1 = fibo(100000)
m = tracemalloc.get_traced_memory()[1]
print(f'内存占用是：{m / 1024 / 1024}MB')

f1 = Fibo(100000)
for n in f1:
    if n > 100:
        break
    print(n)

生成器

 1.生成器函数：函数体中如果出现了 yield 关键字，那该函数是『生成器函数』。
 2.生成器对象：调用『生成器函数』时，其函数体不会立刻执行，而是返回一个『生成器对象』。
备注：不管能否执行到 yield 所在的位置，只要函数中有 yield 关键字，那该函数，就是『生成器函数』。

写在『生成器函数』中的代码，需要通过『生成器对象』来执行：
  1.调用『生成器对象』的 __next__ 方法，会让『生成器函数』中的代码开始执行。
  2.当『生成器函数』中的代码开始执行后，遇到 yield 会“暂停”执行，并且其内部会记录“暂停”的位置。
  3.后续调用 __next__ 方法时，都会从上一次“暂停”的位置，继续运行，直到再次遇到 yield。
  4.遇到 return 会抛出 StopIteration 异常，并将 return 后面的表达式，作为异常信息。
  5.yield 后面所写的表达式，会作为本次 __next__ 方法的返回值。

def demo():
    print('demo函数开始执行了')
    print(100)
    yield '我是第1个yield所返回的数据'
    a = 200
    print(a)
    yield '我是第2个yield所返回的数据'
    b = 300
    print(b)
    return '尚硅谷'

d = demo()
r1 = next(d)
print(r1)
r2 = next(d)
print(r2)
try:
    next(d)
except StopIteration as e:
    print(e)

生成器对象是一种特殊的迭代器（本质是通过 yield 自动实现了迭代器协议）。

def demo():
    print('demo函数开始执行了')
    print(100)
    yield '我是第1个yield所返回的数据'
    a = 200
    print(a)
    yield '我是第2个yield所返回的数据'
    b = 300
    print(b)
    return '尚硅谷'

d = demo()
验证：生成器对象d，和迭代器一样，也拥有：__iter__  和 __next__ 方法
print(hasattr(d, '__iter__'))
print(hasattr(d, '__next__'))

验证：生成器对象的__iter__方法，和迭代器一样，返回的也是自身
result = iter(d)
print(result == d)

for循环遍历生成器
for item in d:
    print(item)

for循环背后的逻辑
gen = iter(d)
while True:
    try:
        value = next(gen)
        print(value)
    except StopIteration:
        break

yield 也能写在循环里

def create_car(total):
    for index in range(1, total + 1):
        yield f'我是第{index}台车'

cars是生成器对象
cars = create_car(5)

调用一次cars的__next__方法，就会得到一台车
c1 = next(cars)
print(c1)
c2 = next(cars)
print(c2)
c3 = next(cars)
print(c3)
c4 = next(cars)
print(c4)
c5 = next(cars)
print(c5)

for car in cars:
    print(car)

yield from 能把一个『可迭代对象』里的东西依次 yield 出去。(替代：for + yield)

def demo():
    nums = [10, 20, 30, 40]
    yield from nums

d = demo()
r1 = next(d)
print(r1)
r2 = next(d)
print(r2)
r3 = next(d)
print(r3)
r4 = next(d)
print(r4)

for item in d:
    print(item)

使用：生成器.send(值) 可以让生成器继续执行的同时，给上一次 yield 传值。
备注1：next 只能取值，send 既能取值，也能送值
备注2：第一次启动生成器，不能传值！

def demo():
    print('demo函数开始执行了')
    print(100)
    a = yield '我是第1个yield所返回的数据'
    print(a)
    b = yield '我是第2个yield所返回的数据'
    print(b)
    return '尚硅谷'

d = demo()
r1 = d.send(None)
print(r1)
r2 = d.send(666)
print(r2)
try:
    d.send(888)
except StopIteration as e:
    print(e)

用生成器实现遍历Person类的实例对象

class Person:
    def __init__(self, name, age, gender, address):
        self.name = name
        self.age = age
        self.gender = gender
        self.address = address
        self.__attr = [name, age, gender, address]

    def __iter__(self):
        # yield self.name
        # yield self.age
        # yield self.gender
        # yield self.address
        yield from self.__attr

p1 = Person('张三', 18, '男', '北京昌平')
# 目标：
for attr in p1:
    print(attr)

用生成器实现斐波那契数列.

def fibo(total):
    pre = 1
    cur = 1

    for index in range(total):
        if index < 2:
            yield 1
        else:
            value = pre + cur
            pre = cur
            cur = value
            yield value

f1 = fibo(10)

for item in f1:
    print(item)

#无论是迭代器，还是生成器对象，都可以用list、tuple、set等直接拿到其里面的所有内容（注意：容易挤爆内存）
result = set(f1)
print(result)

生成器表达式：一种用类似列表推导式的语法，快速创建生成器对象的方式。
语法格式：(表达式 for 变量 in 可迭代对象)
什么时候适合用生成器表达式？———— 当“每个结果，只依赖当前这一个元素”时。

nums = [10, 20, 30, 40]

#列表推导式
result1 = [n * 2 for n in nums]
print(result1)

#生成器表达式
result2 = (n * 2 for n in nums)
for item in result2:
    print(item)

文件操作

介绍

文件的分类：1.纯文本文件、2.二进制文件

纯文本文件：
  1.读取和存储时，要遵循某种『字符编码』规范（如：UTF-8 等）进行编码和解码，最终以『二进制』的形式存储。
  2.『纯文本文件』最终会呈现为：可以直接阅读的文本信息。
  3.常见的『纯文本文件』有：.txt  .py  .md  .html 等。

二进制文件：
  1.读取和存储时，不涉及字符编码，会按照某种『文件格式规范』把内容转为『二进制』数据进行存储。
  2.二进制文件需要由『能够识别其格式的软件』进行解析，最终的呈现形式多种多样（音频、视频、图像、幻灯片等）。
  3.常见的二进制文件有：.mp3  .mp4  .doc  .ppt  .jpg  .png 等。

绝对路径 vs 相对路径：
  1.绝对路径：从文件系统的『根目录』开始，完整描述文件或目录所在的位置。
  2.相对路径：以当前『工作目录』为参照，描述目标文件或目录相对于它的位置。

文件操作流程

操作文件的标准流程：
  1.创建『文件对象』
  2.操作文件（读取、写入 等）
  3.关闭文件

文件操作的核心 ———— open函数：它可以打开或创建文件，且支持多种操作模式，返回值是『文件对象』。
open 函数最常用的三个参数如下：
  1.file：要操作的文件路径
  2.mode：文件的打开模式
      r ：读取（默认值）
      w ：写入，并先截断文件
      x ：排它性创建，如果文件已存在，则创建失败
      a ：打开文件用于写入，如果文件存在，则在文件末尾追加内容
      b ：二进制模式
      t ：文本模式（默认值）
      + ：打开用于更新（读取与写入）
  3.encoding：字符编码

其中r w x a是主模式，不能组合使用，b t +可以组合主模式使用

读取文件

read 方法

read 方法说明：
  1.read(size)中的 size 是可选参数。
     🔸若不传递 size 参数，表示：读取文件中所有的内容（注意内存占用！）。
     🔸若传递了 size 参数，表示：读取文件中指定个数的字符，或指定大小的字节。
  2.read 会从上一次 read 的位置继续读取（指针思想），若到达文件末尾后继续读取，将返回空字符串。

#第一步：创建『文件对象』
file = open(file='a.txt', mode='rt', encoding='utf-8')
file = open('a.txt', 'rt', encoding='utf-8')
file = open('D:/test/atguigu.txt', 'rt', encoding='utf-8')
file = open('D:/test/girl.jpg', 'rb')

#第二步：操作文件（读取）
#多次调用read去逐步读取文件
r1 = file.read(2)
r2 = file.read(3)
r3 = file.read(4)
r4 = file.read()
print(r1, end='')
print(r2, end='')
print(r3, end='')
print(r4, end='')

#用循环配合多次read（对内存友好）
while True:
    result = file.read(10)
    if result == '':
        break
    print(result, end='')

#第三步：关闭文件
file.close()

readline 方法

readline 方法说明：
  1.readline(size) 中的 size 是可选参数。
     🔸若不传递 size 参数，表示：读取当前这一行所有的内容。
     🔸若传递了 size 参数，表示：表示读取当前行时，最多能读取的字符数，或字节数（size不是行数）。
  2.readline 方法，也是从上一次位置继续读取，若到达文件末尾后继续读取，也是返回空字符串。

#第一步：创建『文件对象』
file = open('a.txt', 'rt', encoding='utf-8')

#第二步：操作文件（读取）
#依次调用readline逐行读取
r1 = file.readline()
r2 = file.readline()
r3 = file.readline()
r4 = file.readline()
print(r1.strip())
print(r2.strip())
print(r3.strip())
print(r4.strip())

#通过循环配合readline逐行读取
while True:
    line = file.readline()
    if line == '':
        break
    # print(line.strip())
    print(line, end='')

#第三步：关闭文件
file.close()

使用 for 循环直接遍历文件对象

#第一步：创建『文件对象』
file = open('a.txt', 'rt', encoding='utf-8')

#第二步：操作文件（读取）
for line in file:
    print(line, end='')

#第三步：关闭文件
file.close()

readlines 方法

readlines 方法说明：
  1.readlines(hint) 中的 hint 是可选参数。
      🔸若不传递 hint 参数，表示：读取当前文件的所有行。
      🔸若传递了 hint 参数，表示：期望读取的【字符个数 或 字节数的上限】（hint不是行数）。
  2.注意：由于 readlines 是一次性读取文件的所有内容，所以不适合读取体积较大的文件。

#第一步：创建『文件对象』
file = open('a.txt', 'rt', encoding='utf-8')

#第二步：操作文件（读取）
result = file.readlines()
print(result)

#第三步：关闭文件
file.close()

最佳实践

使用 with 上下文管理器，结合for循环遍历，逐行读取文件。

with open('a.txt', 'rt', encoding='utf-8') as file:
    for line in file:
        print(line, end='')

with说明

1.Python 中的 with 主要用于管理程序中“需要成对出现的操作”，例如：
   🔸上锁 / 解锁
   🔸打开 / 关闭
   🔸借用 / 归还

2.最终目标：编码者只管做具体的事，“进入”和“离开”的事，让 Python 自动处理。

3.语法格式如下：
  with 能得到一个上下文管理器的表达式 as 变量:
      具体的事1
      具体的事2
      具体的事3

4.上下文管理器协议：
   (1). __enter__ 方法：with 中的代码执行【之前】调用，其返回值会赋值给 as 后的变量。
   (2). __exit__  方法：with 中的代码执行【结束后】调用（无论是 with 中否出现异常都会调用）。

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

    def speak(self):
        print(f'我叫{self.name}，年龄是{self.age}')

    def __enter__(self):
        print('-----我是进入的逻辑-----')
        return self

    # 当 with 中的代码发生异常时，__exit__ 方法的返回值规则如下：
    #   🔸返回“真”：表示异常【已经】被处理，异常【不会】被继续抛出。
    #   🔸返回“假”：表示异常【没有】被处理，异常【会】被继续抛出。
    def __exit__(self, exc_type, exc_val, exc_tb):
        print('-----我是离开的逻辑-----')
        # exc_type  : 异常类型
        # exc_val   : 异常对象
        # exc_tb    : 异常追踪信息
        if exc_type:
            print(f'异常类型：{exc_type}')
            print(f'异常对象：{exc_val}')
            print(f'异常追踪信息：{exc_tb}')
        return True

# 1.计算 with 后面的表达式，得到一个『上下文管理器』。
# 2.调用『上下文管理器』的 __enter__() 方法，并将其返回值赋给 as 后面的变量。
# 3.执行 with 所管理的代码。
# 4.无论代 with 中的代码，是正常结束，还是发生异常，都会自动调用『上下文管理器』的 __exit__ 方法。

with Person('张三', 18) as p1:
    p1.speak()
    # p1.study()
    print(666)

写入文件

文件操作的核心 ———— open函数：它可以打开或创建文件，且支持多种操作模式，返回值是『文件对象』。
open 函数最常用的三个参数如下：
  1.file：要操作的文件路径
  2.mode：文件的打开模式
      r ：读取（默认值）
      w ：写入，并先截断文件
      x ：排它性创建，如果文件已存在，则创建失败
      a ：打开文件用于写入，如果文件存在，则在文件末尾追加内容
      b ：二进制模式
      t ：文本模式（默认值）
      + ：打开用于更新（读取与写入）
  3.encoding：字符编码

其中r w x a是主模式，不能组合使用，b t +可以组合主模式使用

测试 w 模式

with open('b.txt', 'wt', encoding='utf-8') as file:
    file.write('你好')

测试 x 模式

with open('demo.txt', 'xt', encoding='utf-8') as file:
    file.write('你好')

测试 a 模式

with open('a.txt', 'at', encoding='utf-8') as file:
    file.write('你好')

缓冲区

在 Python 中文件写入时，并不是每写一次就立刻落盘，而是：先写到“缓冲区”里。

文件对象的 flush 方法：把缓冲区中的数据，立刻写入到文件中。

with open('demo.txt', 'at', encoding='utf-8') as file:
    file.write('你好1')
    file.write('你好2')
    file.flush()
    time.sleep(10000)
    file.write('你好3')
    file.write('你好4')

测试 rt+

with open('a.txt', 'rt+', encoding='utf-8') as file:
    # seek(offset, whence)方法：用于改变文件对象指针的位置，参数说明如下：
    #   offset：偏移量，要移动多少距离
    #   whence：参考点，从哪里开始计算偏移，有三种取值：
    #       0：从文件开头计算（默认值）
    #       1：从当前位置计算
    #       2：从文件末尾计算
    #  注意：在文本模式下，不要随意去定位中文字符位置，否则可能破坏文件编码。
    file.seek(0, 0)
    file.write('你好')

测试 wt+

with open('a.txt', 'wt+', encoding='utf-8') as file:
    file.write('你好')
    file.seek(0, 0)
    result = file.read()
    print(result)

测试 xt+

with open('demo3.txt', 'xt+', encoding='utf-8') as file:
    file.write('你好')
    file.seek(0, 0)
    result = file.read()
    print(result)

测试 at+

with open('a.txt', 'at+', encoding='utf-8') as file:
    file.write('你好')
    file.seek(0, 0)
    result = file.read()
    print(result)

目录操作

import os
import shutil

# os.mkdir(path)：创建“单级”目录（如果目录已经存在，则会抛出异常）
os.mkdir('D:/demo')

# os.makedirs(path)：创建“多级”目录（如果路径中的所有目录都已经存在，则会抛出异常）
os.makedirs('D:/demo/aa/bb')

# os.rmdir(path)：删除空目录（如果目录不存在，或目录非空，都会抛出异常）
os.rmdir('D:/demo/aa/bb')

# os.removedirs(path)：递归删除空目录，在成功删除末尾一级目录后，会“向上”尝试把父级目录也删除（直到父目录不是空目录）
os.removedirs('D:/demo/aa/bb')

# os.path.exists(path)：判断路径是否存在（文件/目录都算）
result = os.path.exists('D:/demo/aa/bb')
print(result)

# os.path.isdir(path)：用于判断路径，具体规则如下：
#   1.路径不存在 ==================> 返回 False
#   2.路径存在，但指向的是文件 =====> 返回 False
#   3.路径存在，并且是目录 =======> 返回 True
result = os.path.isdir('D:/demo/aa/bb')
print(result)

# os.path.isfile(path)：判断是否为文件
result = os.path.isfile('D:/demo/aa/bb')
print(result)

# os.scandir(path)：扫描指定目录
result = os.scandir('D:/demo')
for item in result:
    print('目录' if item.is_dir() else '文件', item.name)

# os.walk(path)：按层级，递归地遍历指定目录下，所有的子目录和文件
result = os.walk('D:/demo')
for item in result:
    print(item)

# ⚠️危险操作：删除有内容的目录
shutil.rmtree('D:/demo')

文件复制

原生复制

def copy_file_original(src, dst):
    with open(src, 'rt',encoding='utf-8') as f1, open(dst, 'wt',encoding='utf-8') as f2:
        while True:
            data = f1.read(1024)
            if not data:
                break
            f2.write(data)
        print("文件复制完毕")

if __name__ == '__main__':
    with open('a.txt','wt',encoding='utf-8') as f1:
        f1.write("你好啊1111")
    copy_file_original('a.txt','b.txt')

工具复制

shutil中工具

import shutil
def copy_file_utils(src, dst):
    shutil.copyfile(src, dst)

if __name__ == '__main__':
    with open('a.txt','wt',encoding='utf-8') as f1:
        f1.write("你好啊1111")
    copy_file_utils('a.txt','c.txt')

进程

概念

一、并发 vs 并行：
    1.并发：
       🔸概念：在一段时间内，当CPU面对多个任务时，会将每个任务交替着执行一段时间。
       🔸特点：
          (1).对于某个瞬间，CPU实际上只在执行一个任务。
          (2).CPU通过高频切换不同的任务，让每个任务都能得到推进，仿佛多个任务在“同时执行”。
    2.并行：
       🔸概念：并行依赖于多个CPU，或多核心的CPU，在同一时刻，每个核心都在执行不同的任务。
       🔸特点：
          (1).对于某个瞬间，每个CPU（或每个核心）都在执行不同的任务。
          (2).通过多个CPU（或多个核心）同时工作的方式，让多个任务真的在同时执行。

  说明：在现代操作系统中，并发与并行通常都是同时存在的。

二、同步 vs 异步：
    1.同步(sync)：
       🔸概念：发起一个任务之后，需要等该任务完成后，才能继续执行后续任务。
       🔸表现：当前执行流会被『阻塞』。
    2.异步(async)：
       🔸概念：发起一个任务之后，不必等该任务完成，就可以继续执行其他任务。
          备注：虽然不必等待任务完成，但任务完成后，仍然可以通过特定方式获取结果。
       🔸表现：当前执行流不会被『阻塞』。

三、注意区分概念：
       并发 / 并行：描述的是任务如何被执行，即：多个任务在执行时，CPU要如何处理。
       同步 / 异步：描述的是任务如何被组织和等待，即：要不要等当前发起的任务执行完，再进行下一个任务。

       ⚠️注意点：CPU 的核心数和执行速度，不会改变任务之间的逻辑依赖关系！例如：
       一旦任务1、任务2、任务3 之间被设计为同步关系，那么：即便 CPU 切换任务的速度再快，核心数量再多，也不会在【任务1】没完成的情况下去启动【任务2】

四、进程与线程
    1.进程：
       (1).一个正在运行的程序或软件，背后就对应着一个或多个进程。
       (2).进程是操作系统进行资源分配的基本单位。
       (3).每个进程都有自己独立的一块内存空间。

    2.线程：
       (1).线程是进程内部的执行单元（一个进程中可以有多个线程）。
       (2).线程是操作系统进行 CPU 调度的基本单位。
       (3).同一进程内的线程共享进程资源。

使用

创建进程

import os
import time
from multiprocessing import Process
print(100, __name__, os.getpid())

# 定义一个 speak 函数，功能是：每隔一秒说话一次（一共说话10次）
def speak():
    for index in range(10):
        print(f'我在说话{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

# 定义一个 study 函数，功能是：每隔一秒学习一次（一共学习15次）
def study():
    for index in range(15):
        print(f'我在学习{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

# 注意：一定要写 if __name__ == '__main__' 这个判断，原因如下：
#   1.当创建子进程时，Python 并不会把父进程内存里的 speak 函数直接交给子进程。
#   2.Python会启动一个全新的 Python 解释器进程，重新执行当前的 .py 文件（作为模块）。
#   3.在执行过程中，重新定义出一个 speak 函数，交给子进程。
if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    # 创建两个 Process 类的实例对象（进程对象），分别是 p1 和 p2。
    # 注意点1：p1 和 p2 就对应着以后的两个子进程，在创建它们的时候，就要指定好他们要执行的任务。
    # 注意点2：此时的 p1 和 p2 只是代码层面的两个进程对象，操作系统还没有真的创建 p1 和 p2 两个进程。
    p1 = Process(target=speak)
    p2 = Process(target=study)

    # 调用进程对象的 start 方法，会立刻向操作系统申请一个进程，并且会将该进程交由操作系统进行调度。
    p1.start()
    p2.start()

    print('我是主进程中的【最后一行】打印')

参数说明

import os
import time
from multiprocessing import Process, current_process

def speak(a, b, msg):
    for index in range(10):
        print(f'{msg}--{a}--{b}--{current_process().name}--我在说话{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

def study():
    for index in range(15):
        print(f'我在学习{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    # Process的参数：
    #   🔸group： 默认值为None（应当始终为None）。
    #   🔸target：子进程要执行的可调用对象，默认值为 None。
    #   🔸name：  进程名称，默认为 None ，如果设置为 None，Python 会自动分配名字。
    #   🔸args：  给 target 传的位置参数（元组）
    #   🔸kwargs：给 target 传的关键字参数（字典）。
    #   🔸daemon：标记进程是否为守护进程，取值为布尔值（默认为 None，表示从创建方进程继承）。
    p1 = Process(target=speak, name='说话进程', args=(666, 888), kwargs={'msg':'尚硅谷'})
    p2 = Process(target=study)
    # print(p1.name)
    # print(p2.name)
    p1.start()
    p2.start()
    print('我是主进程中的【最后一行】打印')

锁

import os
import time
from multiprocessing import Process, Lock,RLock

def speak(lock):
    for index in range(10):
        # 上锁：如果锁是空闲的，立刻上锁，继续往下执行；如果锁被别人拿着：当前进程会阻塞等待（原地等）
        lock.acquire()
        lock.acquire()
        print('好好', end='')
        print('学习', end='')
        print('天天', end='')
        print('向上')
        # 释放锁：acquire 和 release 必须成对出现，否则会永远卡住（死等）
        lock.release()
        lock.release()
        time.sleep(1)


def study(lock):
    for index in range(15):
        # with lock: 会自动完成两件事：
        #   (1).进入前：自动执行 lock.acquire() 上锁
        #   (2).离开后：自动执行 lock.release() 释放锁
        # 好处：即便代码块里发生异常，也能保证释放锁，避免“卡死”
        # 备注：RLock 可以多次上锁
        with lock:
            print('A', end='')
            print('B', end='')
            print('C', end='')
            print('D')
        time.sleep(1)

if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    lock = RLock()
    p1 = Process(target=speak, args=(lock,))
    p2 = Process(target=study, args=(lock,))
    p1.start()
    p2.start()
    print('我是主进程中的【最后一行】打印')

join

join 方法的作用：阻塞当前进程，等 join 前面的进程执行完，再继续往下执行。
join(timeout)，其中 timeout 是可选参数，表示等多久，单位是秒。

注意点：
  1.p.join() 不是让进程 p 等，而是让“执行这行 join 代码的那个进程”去等，等的是 p 这个进程。
  2.当达到了 timeout 所指定的时间后，进程并不会终止，只是“不再等”了。
  3.join 必须在 start 之后

import os
import time
from multiprocessing import Process

def speak():
    for index in range(10):
        print(f'我在说话{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

def study():
    for index in range(15):
        print(f'我在学习{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    p1 = Process(target=speak)
    p2 = Process(target=study)
    p1.start()
    # 让主进程等 p1 5秒钟
    p1.join(5)

    p2.start()
    # p1.join() # 让主进程等 p1 进程执行完毕后，主进程再继续执行。
    # p2.join() # 让主进程等 p2 进程执行完毕后，主进程再继续执行。
    print('我是主进程中的【最后一行】打印')

terminate

import os
import time
from multiprocessing import Process

def speak():
    try:
        for index in range(10):
            print(f'我在说话{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
            time.sleep(1)
    # 注意：使用 terminate 终止进程，不会引起 finally 执行！
    finally:
        print('我是finally里的逻辑')

def study():
    for index in range(15):
        print(f'我在学习{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
        time.sleep(1)

if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    p1 = Process(target=speak)
    p2 = Process(target=study)
    p1.start()
    p2.start()

    time.sleep(3)
    print('我是主进程，我准备强制终止p1进程........')
    # 向操作系统申请强制终止p1进程
    p1.terminate()
    # 等操作系统彻底终止掉了p1进程
    p1.join()
    # 看一下p1进程是否“活着”
    print(p1.is_alive())

    print('我是主进程中的【最后一行】打印')

守护进程

什么是守护进程？
  1.一种 “依附于主进程存在的子进程”，一旦主进程结束，它就会被自动终止。
  2.简言之：主进程一死，守护进程必跟着死。

守护进程的使用场景：
  1.后台监控类任务
  2.日志 / 统计 / 采样 类任务
  3.辅助型“陪跑任务”

注意点：
  1.守护进程必须是 子进程。
  2.主进程结束，守护进程也会随之结束。
  3.守护进程中，不允许再创建新的子进程。
  4.必须在 start 之前，start() 之后，不能再设置 daemon

import os
import time
from multiprocessing import Process

def monitor():
    while True:
        try:
            with open('log.txt', 'r', encoding='utf-8') as file:
                lines = sum(1 for _ in file)
        except FileNotFoundError:
            lines = 0
        print(f'我是【守护进程({os.getpid()})】，log.txt 共有{lines}行')
        time.sleep(1)

def test():
    for index in range(30):
        print(f'我是test({os.getpid()})')
        time.sleep(1)

if __name__ == '__main__':
    print(f'我是主进程({os.getpid()})中的【第一行】代码')

    p1 = Process(target=monitor, daemon=True)
    p2 = Process(target=test)

    p1.start()
    p2.start()

    # 向文件中写入数据
    with open('log.txt', 'a', encoding='utf-8') as file:
        for index in range(10):
            file.write(f'尚硅谷{index}\n')
            file.flush()
            time.sleep(1)

    print(f'我是主进程({os.getpid()})中的【最后一行】代码')

进程之间不共享内存

进程之间不共享内存，因此也就不共享任何变量。

from multiprocessing import Process

def test1(num, names):
    num += 10
    names.append('张三')
    print(f'我是 test1 进程，操作之后的num是{num}，操作之后的names是{names}')

def test2(num, names):
    num -= 10
    names.append('李四')
    print(f'我是 test2 进程，操作之后的num是{num}，操作之后的names是{names}')

if __name__ == "__main__":
    num = 100
    names = []

    print('主进程中的【第一行】代码')
    p1 = Process(target=test1, args=(num, names))
    p2 = Process(target=test2, args=(num, names))

    p1.start()
    p2.start()

    p1.join()
    p2.join()
    print('主进程中的【最后一行】代码', num, names)

Queue

我们之前学过 list、tuple、dict，它们的特点是：数据想怎么拿数据，就怎么拿。
队列(Queue)是：一种“先进先出”的数据结构（先放进去的数据，一定会先取出来）

使用方法

import time
from multiprocessing import Queue, Process

# 创建一个队列（不限制大小，即：不设置容量上限）
# q1 = Queue()

# 创建一个队列（最多能保存3个元素）
# q2 = Queue(3)

# put方法：往队列里放数据（入队）
# q1.put(10)
# q1.put(20)
# q1.put(30)

# get方法：从队列里取数据（出队）
# value1 = q1.get()
# value2 = q1.get()
# value3 = q1.get()
# print(value1)
# print(value2)
# print(value3)

# empty方法：判断队列是否为空
# result = q1.empty()
# print(result)

# full方法：判断队列是否已满
# q1.put(10)
# q1.put(20)
# q1.put(30)
# result = q1.full()
# print(result)

# q2.put(100)
# q2.put(200)
# q2.put(300)
# result = q2.full()
# print(result)

# qsize方法：获取队列长度
# q1.put(10)
# q1.put(20)
# q1.put(30)
# result = q1.qsize()
# print(result)

# 队列具备等待模式
# q2.put(100)
# q2.put(200)
# q2.put(300)

# (1).当队列已满，继续 put，就会进入等待模式，等别人调用get方法，取走一个元素
# q2.put(400)
# print('放入完毕')

# (2).当队列已满，执行：put(元素, timeout=秒数)，就会等待指定秒数。
# q2.put(400, timeout=3)
# print('放入完毕')

# (3).put_nowait 方法，会直接向队列中添加元素，不会进入等待模式，若队列已满，会抛出异常。
# 备注：put_nowait 等价于 put(元素, block=False)，block的默认值为 True
# q2.put_nowait(400)
# q2.put(400, block=False)

# get读多了，也会进入等待模式
# q2.get()
# q2.get()
# q2.get()

# (1).当队列已空，继续 get，就会进入等待模式。x
# q2.get()

# (2).当队列已空，执行 get(timeout=秒数)，就会等待指定秒数。
# q2.get(timeout=3)

# (3).get_nowait 方法，会直接读取队列中的元素，不会进入等待模式，若队列已空，会抛出异常
# 备注：get_nowait 等价于 get(block=False)
# q2.get_nowait()
# q2.get(block=False)

def test(q):
    time.sleep(3)
    result = q.get()
    print('我从队列中取出了一个元素：',result)


# 通过多进程，演示一下：当队列满了以后，再次put会等待，当有人从队列中取出元素后，put会继续。
if __name__ == '__main__':
    # 创建一个队列，让其最多能保存2个元素
    q = Queue(2)
    # put两次，把队列填满
    q.put('尚硅谷')
    q.put('atguigu')
    print(f'队列是否已满：{q.full()}')

    # 创建子进程p1
    p1 = Process(target=test, args=(q, ))
    # 开启子进程p1，让其3秒钟后，从队列中取出一个元素
    p1.start()

    print('即将向已满的队列中添加一个元素........')
    q.put('hello')

    print('目前队列中有的元素是：')
    print(q.get())
    print(q.get())

进程通信

import time
from multiprocessing import Process, Queue

# 子进程1：往队列里放数据
def test1(q):
    for index in range(5):
        print(f'【test1】放入数据：{index}')
        q.put(index)
        time.sleep(0.5)

# 子进程2：从队列里取数据
def test2(q):
    for index in range(5):
        data = q.get()
        print(f'【test2】取出数据：{data}')
        time.sleep(1)

if __name__ == '__main__':
    q = Queue()

    p1 = Process(target=test1, args=(q,))
    p2 = Process(target=test2, args=(q,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

Pipe

import time
from multiprocessing import Process, Pipe

def test1(con1):
    time.sleep(2)
    con1.send(100)
    print('test1发送了100')

def test2(con2):
    data = con2.recv()
    print(f'test2接收了{data}')


if __name__ == '__main__':
    con1, con2 = Pipe(duplex=False)
    p1 = Process(target=test1, args=(con1,))
    p2 = Process(target=test2, args=(con2,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

自定义进程类

继承Process类，重写run方法即可

from multiprocessing import Process
import os, time

class SpeakProcess(Process):
    def __init__(self, a, b, **kwargs):
        super().__init__(**kwargs)
        self.a = a
        self.b = b

    def run(self):
        for index in range(10):
            print(f'{self.a}--{self.b}--我在说话{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
            time.sleep(1)

class StudyProcess(Process):
    def run(self):
        for index in range(15):
            print(f'我在学习{index}, 进程pid是:{os.getpid()}, 我的父进程是:{os.getppid()}')
            time.sleep(1)

if __name__ == '__main__':
    print('我是主进程中的【第一行】打印')
    p1 = SpeakProcess(100, 200)
    p2 = StudyProcess()

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print('我是主进程中的【最后一行】打印')

进程池

创建『进程池执行器』、使用 submit 方法提交任务、使用 shutdown 方法等待任务完成。

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    time.sleep(1)

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    executor = ProcessPoolExecutor(3)
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主进程）
    executor.submit(work, 1)
    executor.submit(work, 2)
    executor.submit(work, 3)
    executor.submit(work, 4)
    executor.submit(work, 5)
    executor.submit(work, 6)
    executor.submit(work, 7)
    # shutdown 的作用：不再接收新的任务。
    # wait=True 的作用：阻塞主进程，等待进程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    print('---------end-------------')

获取子进程执行后的返回结果（Future类的实例对象 + result方法）

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    time.sleep(1)
    return f'我是任务{n}的结果'

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    executor = ProcessPoolExecutor(3)
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主进程）
    # f1 = executor.submit(work, 1)
    # f2 = executor.submit(work, 2)
    # f3 = executor.submit(work, 3)
    # f4 = executor.submit(work, 4)
    # f5 = executor.submit(work, 5)
    # f6 = executor.submit(work, 6)
    # f7 = executor.submit(work, 7)
    futures = [executor.submit(work, index) for index in range(1, 8)]
    # 阻塞主进程，等待进程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    # print(f1.result())
    # print(f2.result())
    # print(f3.result())
    # print(f4.result())
    # print(f5.result())
    # print(f6.result())
    # print(f7.result())
    for f in futures:
        print(f.result())
    print('---------end-------------')

使用 as_completed：按“完成顺序”获取结果

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'我是任务{n}的结果'

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    executor = ProcessPoolExecutor(3)
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主进程）
    futures = [executor.submit(work, index) for index in range(1, 8)]
    # 准备一个 result_list 去收集任务的具体结果
    result_list = []
    # 收集每个任务的结果
    for f in as_completed(futures):
        result_list.append(f.result())
    # 阻塞主进程，等待进程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    # 打印最终的结
    print(result_list)
    print('---------end-------------')

使用 add_done_callback 方法，为任务添加完成时的回调函数

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'我是任务{n}的结果'

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    executor = ProcessPoolExecutor(3)
    # 准备一个 result_list 列表去收集任务的结果
    result_list = []
    # 任务完成后的回调函数
    def done_func(futrue):
        result_list.append(futrue.result())
    # 开启7个任务，并指定回调函数
    for index in range(1, 8):
        f = executor.submit(work, index)
        f.add_done_callback(done_func)
    # 等所有任务都完成
    executor.shutdown(wait=True)
    # 打印最终的结果（按“完成的顺序”获取）
    print(result_list)
    print('---------end-------------')

使用 map 方法批量提交任务（注意：map方法本身不阻塞，但读取其返回的生成器对象是阻塞的，并且得到结果的顺序，与任务分配的顺序是一致的）
换一种说法：map方法会把这一批任务提交到进程池里执行，它会立刻返回一个生成器，真正的阻塞发生在：生成器取结果时（如 list(result)

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'我是任务{n}的结果'

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    executor = ProcessPoolExecutor(3)
    # 通过 map 方法批量提交任务（结果按照提交的顺序来）
    results = executor.map(work, [1, 2, 3, 4, 5, 6, 7])
    # 获取 results 生成器中的内容
    print(list(results))
    # 等所有任务都完成
    executor.shutdown(wait=True)
    print('---------end-------------')

使用 with：进程池的“自动回收”写法，离开 with 代码块时自动执行 shutdown(wait=True)

def work(n):
    print(f'work正在执行任务{n}.........{os.getpid()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'我是任务{n}的结果'

if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个进程池执行器
    with ProcessPoolExecutor(3) as executor:
        # 通过 map 方法批量提交任务（结果按照提交的顺序来）
        results = executor.map(work, [1, 2, 3, 4, 5, 6, 7])
        # 获取 results 生成器中的内容
        print(list(results))
    print('---------end-------------')

线程

线程和进程使用方法一样，引入的包有所区别

如引入线程和线程锁和线程Id如下

from threading import get_native_id, Thread, RLock

使用

import os
import time
from threading import get_native_id, Thread, RLock

def speak(lock):
    for index in range(5):
        with lock:
            print(f'我在说话{index}, 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
        time.sleep(1)

def study(lock):
    for index in range(5):
        with lock:
            print(f'我在学习{index}, 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
        time.sleep(1)

if __name__ == '__main__':
    print(f'-------start------- 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
    lock = RLock()
    # Thread 的参数：
    # 🔸group： 默认值为 None（应当始终为 None）。
    # 🔸target： 子线程要执行的可调用对象，默认值为 None。
    # 🔸name： 线程名称，默认为 None。如果设置为 None，Python 会自动分配名字
    # 🔸args： 给 target 传的位置参数（元组）。
    # 🔸kwargs： 给 target 传的关键字参数（字典）。
    # 🔸daemon： 标记线程是否为守护线程，取值为布尔值（默认为 None）。
    # 使用 Thread 创建线程对象
    t1 = Thread(target=speak, args=(lock,))
    t2 = Thread(target=study, args=(lock,))
    # 调用线程对象的 start 方法，会立刻将该线程交由操作系统进行调度。
    t1.start()
    t2.start()
    # 让主线程等 t1和t2 线程执行完毕后，主线程再继续执行。
    t1.join()
    t2.join()
    print('-------end-------')

继承线程使用

import os
import time
from threading import get_native_id, Thread, RLock

class SpeakThread(Thread):
    def __init__(self, lock, **kwargs):
        super().__init__(**kwargs)
        self.lock = lock
        
    def run(self):
        for index in range(5):
            with self.lock:
                print(f'我在说话{index}, 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
            time.sleep(1)

class StudyThread(Thread):
    def __init__(self, lock, **kwargs):
        super().__init__(**kwargs)
        self.lock = lock

    def run(self):
        for index in range(5):
            with self.lock:
                print(f'我在学习{index}, 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
            time.sleep(1)

if __name__ == '__main__':
    print(f'-------start------- 进程pid是:{os.getpid()}, 线程编号:{get_native_id()}')
    lock = RLock()
    # 继承 Thread 类创建线程对象
    t1 = SpeakThread(lock)
    t2 = StudyThread(lock)
    # 调用线程对象的 start 方法，会立刻将该线程交由操作系统进行调度。
    t1.start()
    t2.start()
    # 让主线程等 t1和t2 线程执行完毕后，主线程再继续执行。
    t1.join()
    t2.join()
    print('-------end-------')

线程池

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import get_native_id, RLock


# 创建『线程池执行器』、使用 submit 方法提交任务、使用 shutdown 方法等待任务完成。
def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    time.sleep(1)


if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个线程池执行器
    executor = ThreadPoolExecutor(3)
    # 创建线程锁
    lock = RLock()
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主线程）
    executor.submit(work, 1, lock)
    executor.submit(work, 2, lock)
    executor.submit(work, 3, lock)
    executor.submit(work, 4, lock)
    executor.submit(work, 5, lock)
    executor.submit(work, 6, lock)
    executor.submit(work, 7, lock)
    # shutdown 的作用：不再接收新的任务。
    # wait=True 的作用：阻塞主线程，等待线程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    print('---------end-------------')


# 获取子线程执行后的返回结果（Future类的实例对象 + result方法）
def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    time.sleep(1)
    return f'任务{n}的结果'


if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个线程池执行器
    executor = ThreadPoolExecutor(3)
    # 创建线程锁
    lock = RLock()
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主线程）
    futures = [executor.submit(work, index, lock) for index in range(1, 8)]
    # 阻塞主线程，等待线程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    # 打印结果
    for f in futures:
        print(f.result())
    print('---------end-------------')


# 使用 as_completed：按“完成顺序”获取结果
def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'任务{n}的结果'


if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个线程池执行器
    executor = ThreadPoolExecutor(3)
    # 创建线程锁
    lock = RLock()
    # 使用 submit 方法提交任务（submit 只负责“提交任务”，不会阻塞主线程）
    futures = [executor.submit(work, index, lock) for index in range(1, 8)]
    # 收集每个线程返回的结果
    result_list = []
    # 将每个线程返回的结果，存入result_list
    for f in as_completed(futures):
        result_list.append(f.result())
    # 阻塞主线程，等待线程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    # 打印最终的结果
    print(result_list)
    print('---------end-------------')


# 使用 add_done_callback 方法，为任务添加完成时的回调函数
def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'任务{n}的结果'


if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个线程池执行器
    executor = ThreadPoolExecutor(3)
    # 创建线程锁
    lock = RLock()
    # 收集每个线程的执行结果
    result_list = []


    # 定义一个线程执行成功后的回调函数
    def done_func(f):
        result_list.append(f.result())


    # 使用submit提交任务，并指定回调函数
    for index in range(1, 8):
        f = executor.submit(work, index, lock)
        f.add_done_callback(done_func)
    # 阻塞主线程，等待线程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    # 打印最终的结果
    print(result_list)
    print('---------end-------------')


# 使用 map 方法批量提交任务（注意：map方法本身不阻塞，但读取其返回的生成器对象是阻塞的，并且得到结果的顺序，与任务分配的顺序是一致的）
#  换一种说法：map方法会把这一批任务提交到线程池里执行，它会立刻返回一个生成器，真正的阻塞发生在：生成器取结果时（如 list(result)

def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'任务{n}的结果'


if __name__ == '__main__':
    print('---------start-------------')
    # 创建一个线程池执行器
    executor = ThreadPoolExecutor(3)
    # 创建线程锁
    lock = RLock()
    # 使用map方法批量提交任务
    result = executor.map(work, [1, 2, 3, 4, 5, 6, 7], [lock] * 7)
    # 打印最终的结果
    print(list(result))
    # 阻塞主线程，等待线程池中所有任务执行完毕。
    executor.shutdown(wait=True)
    print('---------end-------------')


# 使用 with：线程池的“自动回收”写法，离开 with 代码块时自动执行 shutdown(wait=True)
def work(n, lock):
    with lock:
        print(f'work正在执行任务{n}.........{get_native_id()}')
    if n == 1:
        time.sleep(15)
    elif n == 2:
        time.sleep(10)
    else:
        time.sleep(1)
    return f'任务{n}的结果'


if __name__ == '__main__':
    print('---------start-------------')
    with ThreadPoolExecutor(3) as executor:
        # 创建线程锁
        lock = RLock()
        # 使用map方法批量提交任务
        result = executor.map(work, [1, 2, 3, 4, 5, 6, 7], [lock] * 7)
        # 打印最终的结果
        print(list(result))
    print('---------end-------------')

GIL锁（概念性了解即可）

关于GIL（Global Interpreter Lock）
1️、GIL锁是 CPython 解释器中的一把互斥锁。
2️、GIL锁的作用：无论 CPU 有多少个核心，在某一时刻，只允许同一个进程中的一个线程去执行 Python 代码。

结论：CPython 解释器中的多线程模型，本质上是并发，而不是并行！（是快速切换，而不是同时进行）

为何要这样设计？———— 为了确保解释器级别的数据安全。

GIL锁和编码时使用的 Lock 和 Rlock 不是同一个东西。
Lock 和 Rlock是业务层面的锁，目标是：让业务逻辑别出错

Rlock示例1：让打印是完整的

import time
from threading import Thread, RLock,current_thread
def show_info1(lock):
    for index in range(10):
        with lock:
            print('尚', end='')
            print('硅', end='')
            print('谷')

def show_info2(lock):
    for index in range(10):
        with lock:
            print('at', end='')
            print('gui', end='')
            print('gu')

if __name__ == '__main__':
    lock = RLock()
    t1 = Thread(target=show_info1, args=(lock,))
    t2 = Thread(target=show_info2, args=(lock,))
    t1.start()
    t2.start()

Rlock示例2：不要让两个窗口卖出同一张票

current = 1



import time
from threading import Thread, RLock,current_thread
def sale(lock):
    global current
    while True:
        with lock:
            if current <= 20:
                print(f'{current_thread().name}出售了第{current}张票！')
                current += 1
            else:
                print('票已售空')
                break
        time.sleep(0.3)

if __name__ == '__main__':
    lock = RLock()
    t1 = Thread(target=sale, name='窗口1', args=(lock,))
    t2 = Thread(target=sale, name='窗口2', args=(lock,))
    t3 = Thread(target=sale, name='窗口3', args=(lock,))
    t1.start()
    t2.start()
    t3.start()

进程和线程选择

CPU密集型任务，更适合用多进程。

import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

# 准备一个 CPU 密集型任务
def cpu_task(n):
    print(f'任务{n}开始了')
    total = 0
    for i in range(10000000):
        total += i * i
    return total

if __name__ == '_main_':
    print('===== 多进程完成【CPU密集型任务】=====')
    start = time.time()
    #开启四个进程进行计算
    with ProcessPoolExecutor(4) as executor:
        list(executor.map(cpu_task, [1, 2, 3, 4]))
    end = time.time() - start
    print(f'多进程总耗时：{end}秒')


    # print('===== 多线程完成【CPU密集型任务】=====')
    # start = time.time()
    # # 开启四个线程进行计算
    # with ThreadPoolExecutor(max_workers=4) as executor:
    #     results = list(executor.map(cpu_task, [1, 2, 3, 4]))
    # end = time.time() - start
    # print(f'多线程总耗时：{end}秒')

IO密集型任务，更适合用多线程

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# 拷贝文件
def copy_file(index):
    with open('a.zip', 'rb') as src, open(f'a_副本{index}.zip', 'wb') as dst:
        while True:
            data = src.read(1024 * 1024)  # 每次读 1MB
            if not data:
                break
            dst.write(data)

if __name__ == '__main__':
    # print('===== 使用多进程完成【IO密集型任务】 =====')
    # start = time.time()
    # with ProcessPoolExecutor(4) as executor:
    #     for i in range(4):
    #         executor.submit(copy_file, i)
    # end = time.time() - start
    # print(f'多进程耗时：{end} 秒')

    print('===== 使用多线程完成【IO密集型任务】 =====')
    start = time.time()
    with ThreadPoolExecutor(4) as executor:
        for i in range(4):
            executor.submit(copy_file, i)
    end = time.time() - start
    print(f'多线程耗时：{end} 秒')

如果进程和线程耗时差不多优先选择线程，因为线程占用资源少

协程

概念

协程函数，协程对象
1.协程函数(coroutine Function):使用『async关键字』修饰的函数，就是协程函数。
2.协程对象(coroutine Object):调用『协程函数』，就会得到『协程对象』。
注意:调用『协程函数』，并不会执行『协程函数』中的代码。

定义一个协程函数

import asyncio
async def work():
    print("函数执行了")
    return '函数结果'

if __name__ == '__main__':
    fun = work()
#asyncio.run 方法做了3件事:
#  1.创建一个事件循环。
#  2.将收到的协程对象，包装成一个任务(task)，交给事件循环。
#  3.启动事件循环。
# 注意: asyncio.run会阻塞当前线程，直到任务执行完毕，并返回该任务return的最终结果。
    res=asyncio.run(fun)
    print(res)

协程 vs 普通函数

特性	普通函数 (`def`)	协程 (`async def`)
定义方式	`def func()`	`async def func()`
调用方式	`func()`	不能直接调用，必须 `await func()`
执行方式	一次性执行完，遇到 `return` 或结束才返回。	可分步执行，遇到 `await` 会暂停，让出控制权。
返回值	直接返回结果（如 `str`, `int`）。	返回一个协程对象，而不是结果本身。
并发能力	同步执行，后一个任务必须等前一个完成。	异步执行，在等待（如网络I/O）时可切换去执行其他任务。

为什么要用协程？

想象你在厨房做饭：

普通函数（同步）：你必须全程盯着。烧水时，你得等水烧开，然后再去切菜。这期间你什么也做不了，效率很低。
协程（异步）：你不必全程盯着。你烧上水（这是一个耗时操作），在等水开的间隙，你可以去切菜。水烧开了会“通知”你，你再回来处理。这样你就在同一段时间内做了更多的事。

在程序里，协程就扮演了“你”这个角色，await 就是那个“等水开但不用干等着”的指令。

协程的“三步走”运行机制

一个协程从定义到运行，必须经历三个步骤，这也是新手最容易搞混的地方。

第一步：定义协程函数
用 async def 来定义，它本身不会执行任何代码，只是定义了一个蓝图。

async def my_coroutine():
    print("协程开始")
    await asyncio.sleep(1)  # 模拟耗时操作，比如网络请求
    print("协程结束")
    return "完成!"

第二步：获取协程对象
当你调用这个函数时，它不会立即执行，而是返回一个协程对象。

coro_obj = my_coroutine() 
print(coro_obj) # 输出：<coroutine object my_coroutine at 0x...>

第三步：运行协程对象
这是真正让代码跑起来的步骤，有几种方式：

在已有的事件循环中 await：
这是最常见的方式，表示“我等待这个协程执行完毕”。
```
async def main():
    result = await my_coroutine()  # 等待并获取返回值
    print(result)
```
使用 asyncio.run() 启动入口：
这是所有异步程序的“起点”。它会创建一个新的事件循环，并运行你指定的协程直到结束。
```
asyncio.run( main() )
```

创建任务（Task）并发执行：
如果你想“烧水”和“切菜”同时进行，就需要把协程包装成任务。

async def main():
    # 创建任务，让它开始在后台运行
    task1 = asyncio.create_task(my_coroutine())
    task2 = asyncio.create_task(my_coroutine())
    
    # 等待所有任务完成
    await task1
    await task2

await

`await` 到底在等什么？

很多人对 await 的理解就是“等待”，但它的核心机制是“挂起并让出控制权”。

当程序执行到 await some_coroutine() 时：

挂起：当前协程在这里暂停，不继续往下走。
让出：它告诉事件循环：“我要去执行一个耗时操作（比如读数据库），你去忙别的吧，不用等我”。
切换：事件循环收到信号，就会去执行其他准备好的任务（比如另一个协程）。
恢复：等那个耗时操作完成了，事件循环会回来，从刚才挂起的地方继续执行这个协程。

所以，await 是实现“让出控制权”的关键，这也是协程能实现并发的根本原因。

代码理解

理解create_task 与 await 的区别

为了让你更清晰地看到区别，我们来对比两段代码。

场景1：直接 await（串行执行）
两个任务会依次执行，总耗时约 2 秒。

import asyncio
import time

async def say_after(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main():
    start = time.time()
    
    await say_after(1, "你好")
    await say_after(1, "世界")
    
    print(f"总耗时: {time.time() - start:.2f}秒")

asyncio.run(main())
# 输出：你好 (1秒后) 世界 (再1秒后) 总耗时: 2.00秒

场景2：使用 create_task（并发执行）
两个任务会同时开始，总耗时约 1 秒。

async def main():
    start = time.time()
    
    # 创建两个任务，它们会立即开始“准备运行”
    task1 = asyncio.create_task(say_after(1, "你好"))
    task2 = asyncio.create_task(say_after(1, "世界"))
    
    # 等待两个任务都完成
    await task1
    await task2
    
    print(f"总耗时: {time.time() - start:.2f}秒")

asyncio.run(main())
# 输出：你好 (约1秒后同时打印) 世界 总耗时: 1.00秒

总结：协程的关键要点

async def 定义，await 执行：记住，协程函数调用后要拿到结果，必须用 await。
await 不等于 time.sleep()：await 是“让出控制权”，而 time.sleep() 会“阻塞整个线程”，在异步代码中千万别用 time.sleep()，必须用 await asyncio.sleep()。
并发需要 create_task：只有把协程包装成 Task，它们才能“同时”运行。直接 await 是串行的。
事件循环是幕后管理者：asyncio.run() 启动了一个叫“事件循环”的调度员，它负责在协程挂起时，调度其他协程运行。

同步执行多个任务

import asyncio
async def work(n,t):
    print(f"{n} work enter")
    await asyncio.sleep(t)
    print(f"{n} work exit")
    return f'{n}work return'
async def main():
    print('mian enter')
    res1 = await work(1,1)
    print(res1)
    res2 = await work(2,1)
    print(res2)
    res3 = await work(3,1)
    print(res3)
    print('mian exit')
    return 'mian return'
if __name__ == '__main__':
    res=asyncio.run(main())
    print(res)

异步任务执行多任务

import asyncio

import time


async def work(n,t):
    print(f"{n} work enter")
    await asyncio.sleep(t)
    print(f"{n} work exit")
    return f'{n}work return'
async def main():
    start = time.time()
    print('mian enter')
    #asyncio.createtask 会把一个协程对象包装成一个可被事件循环调度的任务，并注册到事件循环中
    task1 = asyncio.create_task(work(1, 1))
    task2 = asyncio.create_task(work(2, 1))
    task3 = asyncio.create_task(work(3, 1))
    res1 = await task1
    print(res1)
    res2 = await task2
    print(res2)
    res3 = await task3
    print(res3)
    print('mian exit')
    end = time.time()
    print('耗时：',end - start)
    return 'mian return'
if __name__ == '__main__':
    res=asyncio.run(main())
    print(res)

升级版，使用asyncio.gather(work(1,2),work(3,2),work(5,2))

import asyncio
import time

async def work(n,t):
    print(f"{n} work enter")
    await asyncio.sleep(t)
    print(f"{n} work exit")
    return f'{n}work return'
async def main():
    start = time.time()
    print('mian enter')
    #把多个协程对象同时丢给事件循环，并在全部执行完后，一次性拿到所有结果。
    res = await asyncio.gather(work(1,2),work(2,2),work(3,2))
    print(res)
    end = time.time()
    print('耗时：',end - start)
    return 'mian return'
if __name__ == '__main__':
    res=asyncio.run(main())
    print(res)

下载图片

传统方式下载

import requests
import time


def download(url, name):
    print(f'开始下载{name}')
    response = requests.get(url)
    print(f'下载完毕{name}')
    with open(name, 'wb') as f:
        f.write(response.content)
def main():
    index = 1
    url_list = [
        'https://img0.baidu.com/it/u=3714152511,1504262391&fm=253&app=138&f=JPEG?w=501&h=500',
        'https://img1.baidu.com/it/u=2996749383,3289036185&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=230933036,341080161&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=3775702182,1659014060&fm=253&app=138&f=JPEG?w=500&h=500'
    ]
    start = time.time()
    for url in url_list:
        index = index + 1
        download(url, str(index)+'.jpg')
    print('耗时=>',time.time() - start)#0.9914
main()

使用协程下载

安装aiohttp

pip install aiohttp



import aiohttp
import time
import asyncio
async def download(url, name,session):
    print(f'开始下载{name}')
    #发送网络请求，获取这张图片，请求发出去后，要等待服务器把数据返回，等的这段时间就是I0等待
    response = await session.get(url)
    print(f'下载完毕{name}')
    #等待数据(图片数据可能分多次传输，需要等待数据全部读完，等的这段时间也是I0等待)
    content = await response.read()
    with open(name, 'wb') as f:
        f.write(content)
    #释放连接资源(告诉aiohttp，这个连接我不用了，你可以回收了)
    await response.release()
async def main():
    url_list = [
        'https://img0.baidu.com/it/u=3714152511,1504262391&fm=253&app=138&f=JPEG?w=501&h=500',
        'https://img1.baidu.com/it/u=2996749383,3289036185&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=230933036,341080161&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=3775702182,1659014060&fm=253&app=138&f=JPEG?w=500&h=500'
    ]
    start = time.time()
    #创建会话
    session = aiohttp.ClientSession()
    #构建协程任务
    req_list = [download(url_list[index],str(index)+'.jpg',session) for index in range(len(url_list))]
    #解包参数
    await asyncio.gather(*req_list)
    print('耗时=>',time.time() - start)#0.9914
    await session.close()

asyncio.run(main()) #0.1365

多线程下载

import time
from concurrent.futures import ThreadPoolExecutor

import requests


def download(url, name):
    print(f'开始下载{name}')
    response = requests.get(url)
    print(f'下载完毕{name}')
    with open(name, 'wb') as f:
        f.write(response.content)


def main():
    index = 1
    url_list = [
        'https://img0.baidu.com/it/u=3714152511,1504262391&fm=253&app=138&f=JPEG?w=501&h=500',
        'https://img1.baidu.com/it/u=2996749383,3289036185&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=230933036,341080161&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=3775702182,1659014060&fm=253&app=138&f=JPEG?w=500&h=500'
    ]
    with ThreadPoolExecutor(4) as exec:
        start = time.time()
        exec.map(download, url_list, ['1.jpg', '2.jpg', '3.jpg', '4.jpg'])
        exec.shutdown()
        print('耗时=>', time.time() - start)  # 0.0.1513


if __name__ == '__main__':
    main()

多进程下载

import time
from concurrent.futures import ProcessPoolExecutor

import requests


def download(url, name):
    print(f'开始下载{name}')
    response = requests.get(url)
    print(f'下载完毕{name}')
    with open(name, 'wb') as f:
        f.write(response.content)


def main():
    index = 1
    url_list = [
        'https://img0.baidu.com/it/u=3714152511,1504262391&fm=253&app=138&f=JPEG?w=501&h=500',
        'https://img1.baidu.com/it/u=2996749383,3289036185&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=230933036,341080161&fm=253&app=138&f=JPEG?w=500&h=500',
        'https://img1.baidu.com/it/u=3775702182,1659014060&fm=253&app=138&f=JPEG?w=500&h=500'
    ]
    with ProcessPoolExecutor(4) as exec:
        start = time.time()
        exec.map(download, url_list, ['1.jpg', '2.jpg', '3.jpg', '4.jpg'])
        exec.shutdown()
        print('耗时=>', time.time() - start)  # 0.0.6425


if __name__ == '__main__':
    main()

调用mysql

安装依赖

数据库依赖

pip install pymysql

连接池

pip install DBUtils

建立数据库连接

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',        # 数据库地址
    port=3306,               # 端口，默认3306
    user='root',             # 用户名
    password='123456',       # 密码
    database='test_db',      # 数据库名
    charset='utf8mb4'        # 字符集，支持emoji
)

# 创建游标（用于执行SQL）
cursor = conn.cursor()

执行 SQL 操作

创建表

create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age INT,
    email VARCHAR(100),
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
cursor.execute(create_table_sql)
conn.commit()  # DDL语句也需要提交

插入数据

# 单条插入
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
cursor.execute(sql, ('张三', 25, 'zhangsan@example.com'))
conn.commit()
print(f"插入成功，ID: {cursor.lastrowid}")

# 批量插入
data = [
    ('李四', 30, 'lisi@example.com'),
    ('王五', 28, 'wangwu@example.com'),
]
cursor.executemany(sql, data)
conn.commit()
print(f"批量插入 {cursor.rowcount} 条")

查询数据

# 查询所有
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)  # 元组形式：(1, '张三', 25, 'zhangsan@example.com', datetime...)

# 查询单条
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
row = cursor.fetchone()
print(row)

# 条件查询
cursor.execute("SELECT * FROM users WHERE age > %s", (20,))
rows = cursor.fetchall()

更新数据

sql = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(sql, (26, '张三'))
conn.commit()
print(f"更新了 {cursor.rowcount} 条记录")

删除数据

sql = "DELETE FROM users WHERE name = %s"
cursor.execute(sql, ('王五',))
conn.commit()
print(f"删除了 {cursor.rowcount} 条记录")

使用字典游标（返回字典而非元组）

import pymysql.cursors

# 使用 DictCursor
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='123456',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor  # 关键参数
)

cursor = conn.cursor()
cursor.execute("SELECT id, name, age FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row['name'])  # 通过字段名访问

事务管理

try:
    conn.begin()  # 开启事务（或直接用 conn.autocommit(False)）
    
    cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ('赵六', 22))
    cursor.execute("UPDATE users SET age = age + 1 WHERE name = %s", ('张三',))
    
    conn.commit()  # 提交事务
    print("事务提交成功")
except Exception as e:
    conn.rollback()  # 发生错误时回滚
    print(f"事务回滚: {e}")
finally:
    cursor.close()
    conn.close()

使用上下文管理器（推荐）

import pymysql

# 自动管理连接和游标
with pymysql.connect(
    host='localhost',
    user='root',
    password='123456',
    database='test_db',
    charset='utf8mb4'
) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        rows = cursor.fetchall()
        for row in rows:
            print(row)
# 退出 with 块时自动关闭游标和连接

使用连接池（生产环境推荐）

from DBUtils.PooledDB import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,  # 使用PyMySQL
    maxconnections=10,  # 最大连接数
    mincached=2,        # 初始空闲连接数
    blocking=True,      # 连接数满时等待
    host='localhost',
    user='root',
    password='123456',
    database='test_db',
    charset='utf8mb4'
)

# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
print(cursor.fetchall())
cursor.close()
conn.close()  # 归还连接到池

防止 SQL 注入

始终使用参数化查询，不要拼接字符串：

# ❌ 危险：SQL注入风险
user_input = "'; DROP TABLE users; --"
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# ✅ 安全：使用参数化
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

常用 API 总结

方法	说明
`cursor.execute(sql, params)`	执行单条 SQL
`cursor.executemany(sql, params_list)`	批量执行
`cursor.fetchone()`	获取一条结果
`cursor.fetchall()`	获取所有结果
`cursor.fetchmany(size)`	获取指定条数
`cursor.lastrowid`	最后插入行的 ID
`cursor.rowcount`	受影响的行数
`conn.commit()`	提交事务
`conn.rollback()`	回滚事务
`conn.close()`	关闭连接

完整示例

import pymysql

class MySQLHelper:
    def __init__(self, host, user, password, database):
        self.conn = pymysql.connect(
            host=host, user=user, password=password,
            database=database, charset='utf8mb4'
        )
        self.cursor = self.conn.cursor()
    
    def insert(self, sql, params):
        self.cursor.execute(sql, params)
        self.conn.commit()
        return self.cursor.lastrowid
    
    def query_all(self, sql, params=None):
        self.cursor.execute(sql, params or ())
        return self.cursor.fetchall()
    
    def update(self, sql, params):
        self.cursor.execute(sql, params)
        self.conn.commit()
        return self.cursor.rowcount
    
    def close(self):
        self.cursor.close()
        self.conn.close()

# 使用
db = MySQLHelper('localhost', 'root', '123456', 'test_db')
db.insert("INSERT INTO users (name, age) VALUES (%s, %s)", ('测试', 20))
rows = db.query_all("SELECT * FROM users")
print(rows)
db.close()

作为服务器

安装依赖

pip install fastapi uvicorn[standard]  # 核心框架 + 高性能服务器

基础示例（Hello World）

# main.py
from fastapi import FastAPI

app = FastAPI(title="我的API服务", version="1.0.0")

@app.get("/")
def root():
    return {"message": "Hello World"}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id, "name": f"用户{user_id}"}

@app.post("/items")
def create_item(name: str, price: float):
    return {"name": name, "price": price, "status": "created"}

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8080 --reload

访问 http://localhost:8080/docs 即可看到自动生成的 Swagger 文档。

龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

大模型安全实战】智能体安全的攻防二元性：同一种能力，为什么既是防线也是攻击面？（第4期）

龙虾开发者社区

SparkSQL 之 Hive On Spark 原理分析

龙虾开发者社区

2026大厂AI Agent面试深度复盘：从“会背概念“到“能扛生产“的范式跃迁

2026年AI Agent面试核心要点摘要（147字）大厂AI Agent面试已从概念背诵转向工程实践考核，重点考察：1）理解Agent与Chatbot的本质差异在于闭环行动能力；2）掌握ReAct循环、记忆分层设计（工作/短期/长期记忆）及优化策略；3）熟悉MCP协议对工具调用的标准化改进及安全风险；4）具备异常处理、性能优化的工程能力。面试官更关注候选人在非确定性系统中构建确定性保障的能力，