Python 高级编程与实战：深入理解面向对象与并发编程-CFANZ编程社区

在上一篇中，我们介绍了 Python 的基础语法和简单的实战项目。本文将深入探讨 Python 的高级特性，包括面向对象编程、异常处理、并发编程等，并通过实战项目帮助你掌握这些高级技术。

1. 面向对象编程进阶

1.1 继承与多态

继承是面向对象编程的重要特性，允许子类继承父类的属性和方法。多态则允许子类重写父类的方法，实现不同的行为。

# 父类
class Animal:
    def speak(self):
        return "Animal sound"

# 子类
class Dog(Animal):
    def speak(self):
        return "Woof!"

class Cat(Animal):
    def speak(self):
        return "Meow!"

# 多态
animals = [Dog(), Cat()]
for animal in animals:
    print(animal.speak())
# 输出:
# Woof!
# Meow!

1.2 类的特殊方法

Python 提供了许多特殊方法（魔术方法），用于定义类的行为。例如，__init__ 用于初始化对象，__str__ 用于定义对象的字符串表示。

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

    def __str__(self):
        return f"{self.name}, {self.age} years old"

# 创建对象
person = Person("Alice", 25)
print(person)  # 输出: Alice, 25 years old

1.3 类属性与实例属性

类属性是类的所有实例共享的属性，而实例属性是每个实例独有的属性。

class MyClass:
    class_attr = "I am a class attribute"

    def __init__(self, instance_attr):
        self.instance_attr = instance_attr

# 访问类属性
print(MyClass.class_attr)  # 输出: I am a class attribute

# 访问实例属性
obj = MyClass("I am an instance attribute")
print(obj.instance_attr)  # 输出: I am an instance attribute

2. 异常处理

异常处理是处理程序运行时错误的重要机制。Python 使用 try-except 块来捕获和处理异常。

# 捕获异常
try:
    result = 10 / 0
except ZeroDivisionError:
    print("Error: Division by zero")
# 输出: Error: Division by zero

# 捕获多个异常
try:
    value = int("abc")
except ValueError:
    print("Error: Invalid integer")
except Exception as e:
    print(f"Unexpected error: {e}")
# 输出: Error: Invalid integer

# finally 块
try:
    file = open("example.txt", "r")
    content = file.read()
except FileNotFoundError:
    print("Error: File not found")
finally:
    file.close()

3. 并发编程

Python 提供了多种并发编程的方式，包括多线程、多进程和异步编程。

3.1 多线程

使用 threading 模块创建和管理线程。

import threading
import time

def worker():
    print("Worker thread started")
    time.sleep(2)
    print("Worker thread finished")

# 创建线程
thread = threading.Thread(target=worker)
thread.start()

# 主线程继续执行
print("Main thread continues")
thread.join()
print("Main thread finished")
# 输出:
# Worker thread started
# Main thread continues
# Worker thread finished
# Main thread finished

3.2 多进程

使用 multiprocessing 模块创建和管理进程。

import multiprocessing
import time

def worker():
    print("Worker process started")
    time.sleep(2)
    print("Worker process finished")

# 创建进程
process = multiprocessing.Process(target=worker)
process.start()

# 主进程继续执行
print("Main process continues")
process.join()
print("Main process finished")
# 输出:
# Worker process started
# Main process continues
# Worker process finished
# Main process finished

3.3 异步编程

使用 asyncio 模块进行异步编程。

import asyncio

async def worker():
    print("Worker coroutine started")
    await asyncio.sleep(2)
    print("Worker coroutine finished")

async def main():
    print("Main coroutine started")
    await worker()
    print("Main coroutine finished")

# 运行事件循环
asyncio.run(main())
# 输出:
# Main coroutine started
# Worker coroutine started
# Worker coroutine finished
# Main coroutine finished

4. Python 实战项目

4.1 简单的聊天服务器

我们将实现一个简单的聊天服务器，使用 socket 模块进行网络通信。

import socket
import threading

def handle_client(client_socket):
    while True:
        message = client_socket.recv(1024).decode('utf-8')
        if not message:
            break
        print(f"Received: {message}")
        client_socket.send(f"Echo: {message}".encode('utf-8'))
    client_socket.close()

def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(('0.0.0.0', 9999))
    server.listen(5)
    print("Server listening on port 9999")

    while True:
        client_socket, addr = server.accept()
        print(f"Accepted connection from {addr}")
        client_thread = threading.Thread(target=handle_client, args=(client_socket,))
        client_thread.start()

# 启动服务器
start_server()

4.2 异步文件下载器

我们将使用 aiohttp 和 asyncio 实现一个异步文件下载器。

import aiohttp
import asyncio

async def download_file(url, filename):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            with open(filename, 'wb') as file:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    file.write(chunk)
            print(f"Downloaded {filename}")

async def main():
    urls = [
        ("https://example.com/file1.txt", "file1.txt"),
        ("https://example.com/file2.txt", "file2.txt")
    ]
    tasks = [download_file(url, filename) for url, filename in urls]
    await asyncio.gather(*tasks)

# 运行事件循环
asyncio.run(main())

5. 总结

本文深入探讨了 Python 的高级特性，包括面向对象编程、异常处理、并发编程等，并通过实战项目帮助你掌握这些技术。通过本文的学习，你应该能够编写更加复杂和高效的 Python 程序。

6. 进一步学习资源

• Python 官方文档 • Python 高级编程 - Real Python • Python 并发编程 - O'Reilly

希望本文能够帮助你进一步提升 Python 编程技能，祝你在编程的世界中不断进步！