并发编程与异步 I/O

并发编程与异步 I/O

1. 并发编程基础

1.1. 并发 vs 并行

  • 并发:多个任务可能在同一时间段内进行,但未必同时在执行。
  • 并行:多个任务在同一时刻真正地同时执行。

1.2. Python 中的并发编程

Python 提供了多种方法实现并发,包括:

  • threading 模块(基于线程)
  • multiprocessing 模块(基于进程)
  • asyncio 模块(基于异步 I/O)

2. 使用 threading 模块

2.1. 创建线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import threading
import time

def worker():
print("Worker is starting...")
time.sleep(2)
print("Worker is done.")

thread = threading.Thread(target=worker)
thread.start()

print("Main thread continues...")
thread.join()
print("Main thread finishes.")

2.2. 使用线程池

1
2
3
4
5
6
7
8
9
10
from concurrent.futures import ThreadPoolExecutor

def task(n):
print(f"Task {n} is starting...")
time.sleep(1)
print(f"Task {n} is done.")

with ThreadPoolExecutor(max_workers=3) as executor:
for i in range(5):
executor.submit(task, i)

3. 使用 multiprocessing 模块

3.1. 创建进程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import multiprocessing
import time

def worker(number):
print(f"Worker {number} is starting...")
time.sleep(2)
print(f"Worker {number} is done.")

if __name__ == '__main__':
processes = []
for i in range(5):
p = multiprocessing.Process(target=worker, args=(i,))
processes.append(p)
p.start()

for p in processes:
p.join()

print("All workers are done.")

3.2. 使用进程池

1
2
3
4
5
6
7
8
9
from multiprocessing import Pool

def square(n):
return n * n

if __name__ == '__main__':
with Pool(processes=4) as pool:
results = pool.map(square, range(10))
print(results) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

4. 异步 I/O 使用 asyncio

4.1. 异步函数和事件循环

1
2
3
4
5
6
7
8
import asyncio

async def main():
print("Main function is starting...")
await asyncio.sleep(2)
print("Main function is done.")

asyncio.run(main())

4.2. 任务并发执行

1
2
3
4
5
6
7
8
9
10
11
12
import asyncio

async def fetch_data(num):
print(f"Fetching data {num}...")
await asyncio.sleep(1)
print(f"Data {num} fetched.")

async def main():
tasks = [fetch_data(i) for i in range(5)]
await asyncio.gather(*tasks)

asyncio.run(main())

4.3. 异步 I/O 结合网络请求

使用 aiohttp 模块进行异步 HTTP 请求:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import aiohttp
import asyncio

async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()

async def main():
urls = ['http://example.com'] * 5
tasks = [fetch(url) for url in urls]
responses = await asyncio.gather(*tasks)
for idx, response in enumerate(responses):
print(f"Response from request {idx+1}: {len(response)} characters.")

asyncio.run(main())

5. 总结

  • 并发编程是提高程序效率的有效手段。
  • Python 提供了多种并发编程的方式,适用于不同的应用场景。
  • 通过掌握 threadingmultiprocessingasyncio,可以熟练地处理多任务问题,提高程序的性能。

以上是关于并发编程与异步 I/O 的基础教程,掌握这些内容有助于进行更复杂的 Python 开发。

元组的定义与使用

元组的定义与使用

1. 什么是元组

元组(tuple)是Python中的一种内置数据类型,用于存储多个元素。元组与列表相似,但与列表不同的是,元组是不可变的,这意味着一旦创建,元组的内容就不能更改。

2. 元组的定义

元组通过一对小括号 () 来定义,元素之间用逗号 , 分隔。

1
2
3
# 示例:定义一个元组
my_tuple = (1, 2, 3)
print(my_tuple) # 输出: (1, 2, 3)

2.1 单元素元组

如果你想定义一个只包含一个元素的元组,需要在元素后面加上一个逗号 ,,否则Python会将其视为该元素的普通数据类型。

1
2
3
# 示例:定义一个单元素元组
single_element_tuple = (1,)
print(single_element_tuple) # 输出: (1,)

3. 元组的访问

可以通过索引来访问元组中的元素,索引从 0 开始。

1
2
3
4
5
# 示例:访问元组中的元素
my_tuple = (10, 20, 30)
print(my_tuple[0]) # 输出: 10
print(my_tuple[1]) # 输出: 20
print(my_tuple[2]) # 输出: 30

4. 元组的切片

元组支持切片操作,语法与列表相同,可以通过 : 来获取元组的一个切片。

1
2
3
4
# 示例:元组切片
my_tuple = (1, 2, 3, 4, 5)
slice_tuple = my_tuple[1:4] # 获取索引1到3的元素
print(slice_tuple) # 输出: (2, 3, 4)

5. 元组的长度

使用 len() 函数可以获取元组的长度(元素个数)。

1
2
3
4
# 示例:获取元组的长度
my_tuple = (1, 2, 3, 4, 5)
length = len(my_tuple)
print(length) # 输出: 5

6. 元组的连接与重复

可以使用 + 运算符连接两个元组,使用 * 运算符重复元组。

1
2
3
4
5
6
7
8
9
10
11
# 示例:元组的连接与重复
tuple_a = (1, 2, 3)
tuple_b = (4, 5, 6)

# 连接元组
combined_tuple = tuple_a + tuple_b
print(combined_tuple) # 输出: (1, 2, 3, 4, 5, 6)

# 重复元组
repeated_tuple = tuple_a * 2
print(repeated_tuple) # 输出: (1, 2, 3, 1, 2, 3)

7. 元组的解包

元组可以通过解包将其元素赋值给多个变量。

1
2
3
4
5
6
# 示例:元组解包
my_tuple = (10, 20, 30)
a, b, c = my_tuple
print(a) # 输出: 10
print(b) # 输出: 20
print(c) # 输出: 30

8. 元组的嵌套

元组可以嵌套,即元组的元素可以是另一个元组。

1
2
3
4
# 示例:元组的嵌套
nested_tuple = ((1, 2), (3, 4), (5, 6))
print(nested_tuple[0]) # 输出: (1, 2)
print(nested_tuple[1]) # 输出: (3, 4)

9. 元组的优点与缺点

9.1 优点

  • 不可变性:元组的不可变性使其可以作为字典的键或集合的元素。
  • 性能:元组有较高的性能,特别是在存储一个确定数量的元素时。

9.2 缺点

  • 不可修改:元组一旦创建,无法更改内容,较少使用的场景。

10. 结语

元组是Python中一个非常有用的基础数据类型,了解元组的定义和使用,可以帮助你更好地处理数据结构。在实际编程中,合理选用列表和元组将带来更好的代码质量和性能表现。

单元测试基础

单元测试基础

在软件开发中,单元测试是确保代码质量的重要手段。Python 提供了一个内置的unittest模块,可以方便地编写和运行测试。以下是关于单元测试基础的详细介绍。

1. 什么是单元测试

单元测试是对软件中最小可测试单元(通常是函数或类)的检验。通过编写单元测试,可以确保代码在修改后仍然按预期工作。

2. unittest模块介绍

unittest是Python标准库中的一个模块,用于创建和运行单元测试。它提供了一套丰富的工具来帮助你测试代码:

  • 测试用例:继承自unittest.TestCase的类,用于编写具体的测试。
  • 测试夹具setUp()tearDown()方法,可以在测试前后执行准备和清理工作。
  • 断言:用于验证测试结果的各种方法,如assertEqual()assertTrue()等。

3. 创建第一个单元测试

3.1 示例代码

假设我们有一个简单的加法函数,如下所示:

1
2
def add(a, b):
return a + b

3.2 编写测试用例

我们可以为这个加法函数编写一个测试用例:

1
2
3
4
5
6
7
8
9
10
11
12
import unittest

class TestAddFunction(unittest.TestCase):

def test_add_positive_numbers(self):
self.assertEqual(add(1, 2), 3)

def test_add_negative_numbers(self):
self.assertEqual(add(-1, -1), -2)

def test_add_zero(self):
self.assertEqual(add(0, 0), 0)

3.3 运行测试

在运行测试之前,我们需要确保代码在同一文件中或者通过导入来使用。运行测试的方式可以在命令行中使用以下命令:

1
python -m unittest discover

4. 断言方法

unittest提供了多种断言方法,以下是一些常用的断言方法:

  • assertEqual(a, b): 断言ab相等。
  • assertNotEqual(a, b): 断言ab不相等。
  • assertTrue(x): 断言x为真。
  • assertFalse(x): 断言x为假。
  • assertIsNone(x): 断言xNone
  • assertIsInstance(a, b): 断言ab的实例。

5. 使用测试夹具

有时候,测试需要一些准备工作,比如创建对象或配置环境。在这种情况可以使用setUp()tearDown()方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
class TestAddFunction(unittest.TestCase):

def setUp(self):
# 在每个测试之前执行
self.a = 3
self.b = 5

def tearDown(self):
# 在每个测试之后执行
pass

def test_add(self):
self.assertEqual(add(self.a, self.b), 8)

6. 运行单个测试

如果只想运行某个特定的测试用例,可以在命令行中指定测试方法:

1
python -m unittest TestAddFunction.test_add_positive_numbers

7. 测试异常

我们还可以测试代码是否抛出预期的异常。可以使用assertRaises方法:

1
2
3
4
5
6
7
8
def divide(x, y):
return x / y

class TestDivideFunction(unittest.TestCase):

def test_divide_by_zero(self):
with self.assertRaises(ZeroDivisionError):
divide(1, 0)

8. 总结

通过使用unittest模块,Python开发者可以轻松编写健壮的单元测试,从而提高代码的可靠性和可维护性。在实际项目中,建议您将测试与代码一起管理,以便在代码更改后快速验证其正确性。

更多的单元测试内容,您可以参考官方文档或其他进阶书籍,例如《Python Testing with unittest, nose, and pytest》。