当前位置：首页 > news >正文

Python常见面试题的详解14

news 来源：原创 2025/5/14 11:18:58

1. 从变量 A 中匹配 JSON 字符串

在处理文本数据时，有时需要从变量里找出其中的 JSON 字符串。JSON 字符串一般以 { 或 [ 开头，以 } 或 ] 结尾，但简单的正则匹配可能不够严谨，所以还需用 json 模块进行验证。

要点

利用正则表达式 \{.*?\} 和 \[.*?\] 分别匹配可能的 JSON 对象和数组字符串。
使用 re.findall 函数找出所有可能的匹配结果。
借助 json.loads 函数对匹配结果进行验证，只有能成功解析的才认定为有效的 JSON 字符串。
若处理的文本包含复杂的嵌套 JSON 结构，简单的正则表达式可能无法准确匹配。此时，可以考虑编写更复杂的正则表达式，或者使用递归方法来处理嵌套结构。另外，对于包含大量数据的文本，频繁调用 json.loads 会影响性能，可以先通过正则表达式筛选出更可能是 JSON 的字符串，再进行验证。

示例

python

import re
import json

# 示例变量 A，包含多种文本和可能的 JSON 字符串
A = 'Some text before {"name": "John", "age": 30} and some text after [1, 2, 3] more text'
# 定义正则表达式模式，用于匹配可能的 JSON 对象和数组
patterns = [r'\{.*?\}', r'\[.*?\]']
json_strings = []

# 遍历模式列表，查找所有可能的匹配项
for pattern in patterns:
    matches = re.findall(pattern, A)
    for match in matches:
        try:
            # 尝试将匹配结果解析为 JSON
            json.loads(match)
            json_strings.append(match)
        except json.JSONDecodeError:
            # 若解析失败，说明不是有效的 JSON 字符串，跳过
            continue

print(json_strings)

2. 过滤评论中的表情

在处理评论数据时，表情符号可能会干扰分析过程，因此需要将其过滤掉。表情符号通常对应特定的 Unicode 编码范围，可利用正则表达式匹配并去除。

要点

定义一个正则表达式，涵盖常见表情符号的 Unicode 编码范围。
使用 re.sub 函数将匹配到的表情符号替换为空字符串，实现过滤。
随着新的表情符号不断出现，需要定期更新 Unicode 范围。此外，对于一些特殊字体或变体的表情符号，可能需要进一步调整正则表达式。还可以考虑结合机器学习方法，通过训练模型来识别和过滤表情符号，提高过滤的准确性。

示例

python

import re

def filter_emoji(comment):
    # 定义正则表达式，匹配常见表情符号的 Unicode 范围
    emoji_pattern = re.compile("["
                               u"\U0001F600-\U0001F64F"  # 表情符号
                               u"\U0001F300-\U0001F5FF"  # 符号与象形图
                               u"\U0001F680-\U0001F6FF"  # 交通与地图符号
                               u"\U0001F1E0-\U0001F1FF"  # 旗帜（iOS）
                               u"\U00002702-\U000027B0"
                               u"\U000024C2-\U0001F251"
                               "]+", flags=re.UNICODE)
    # 将匹配到的表情符号替换为空字符串
    return emoji_pattern.sub(r'', comment)

# 示例评论，包含表情符号
comment = "这条评论有😀表情哦👍"
filtered_comment = filter_emoji(comment)
print(filtered_comment)

3. `search` 和 `match` 的区别

re.search 和 re.match 是 re 模块中用于正则匹配的重要函数，它们的主要差异在于匹配的起始位置。

要点

re.match 函数从字符串的开头开始匹配，如果开头不匹配则返回 None。
re.search 函数会在整个字符串中查找匹配的位置，只要字符串中有一处匹配就会返回匹配对象。
当需要确保匹配从字符串开头开始时，应使用 match 函数；当只关心字符串中是否存在匹配，而不关心位置时，使用 search 函数更合适。此外，match 和 search 返回的匹配对象都有一些方法和属性，如 group() 用于获取匹配的字符串，start() 和 end() 用于获取匹配的起始和结束位置。

示例

python

import re

text = "Hello, World!"
pattern = r"World"

# 使用 match 函数进行匹配
match_result = re.match(pattern, text)
if match_result:
    print("match 匹配成功")
else:
    print("match 匹配失败")

# 使用 search 函数进行匹配
search_result = re.search(pattern, text)
if search_result:
    print("search 匹配成功")
else:
    print("search 匹配失败")

4. 进程总结

进程是程序在操作系统中的一次执行过程，是系统进行资源分配和调度的基本单位。每个进程都拥有独立的内存空间和系统资源，如文件描述符等。

要点

进程是程序执行的实例，有自己独立的资源。
在 Python 中，使用 multiprocessing 模块创建和管理进程。
Process 类用于创建新的进程，start 方法启动进程，join 方法等待进程结束。
进程间通信（IPC）是多进程编程中的重要环节，常见的 IPC 机制包括管道（Pipe）、队列（Queue）和共享内存等。不同的 IPC 机制适用于不同的场景，如管道适用于简单的单向或双向通信，队列适用于多个进程间的数据共享和同步。

示例

python

import multiprocessing

def worker():
    print("子进程执行")

if __name__ == '__main__':
    # 创建一个新的进程
    p = multiprocessing.Process(target=worker)
    # 启动进程
    p.start()
    # 等待进程执行完毕
    p.join()

5. 多进程、多线程和协程的理解与应用

多进程：多个进程可以在多核 CPU 上并行执行，每个进程有独立的内存空间和系统资源。适用于计算密集型任务，如大规模数据计算、图像处理等。
多线程：一个进程可以包含多个线程，线程共享进程的内存空间和系统资源。适合 I/O 密集型任务，如网络请求、文件读写等。
协程：协程是一种轻量级的线程，由程序自身控制执行流程，不需要操作系统进行调度。适合处理大量的 I/O 密集型任务，性能较高。
要点

多进程利用多核 CPU 实现并行计算，但创建和销毁开销大，进程间通信复杂。
多线程共享资源，适合 I/O 密集型任务，但可能存在线程安全问题。
协程轻量级，由程序控制调度，适合大量 I/O 任务，性能高。
在实际项目中，可以根据任务的特点综合使用多进程、多线程和协程。例如，在一个网络爬虫项目中，可以使用多进程充分利用多核 CPU，每个进程中使用多线程或协程处理多个网络请求，提高爬虫的效率。

示例

python

# 多进程示例
import multiprocessing

def process_worker():
    print("多进程工作中")

if __name__ == '__main__':
    p = multiprocessing.Process(target=process_worker)
    p.start()
    p.join()

# 多线程示例
import threading

def thread_worker():
    print("多线程工作中")

t = threading.Thread(target=thread_worker)
t.start()
t.join()

# 协程示例
import asyncio

async def coroutine_worker():
    print("协程工作中")

asyncio.run(coroutine_worker())

6. 说明异步使用场景

Python 的异步编程可以提高程序的并发性能，尤其适用于 I/O 密集型任务。

要点

网络编程中，异步编程可在等待网络响应时处理其他任务，提高并发性能。
文件读写和数据库操作等 I/O 密集型任务也适合使用异步编程，避免阻塞线程。
除了 asyncio 和 aiohttp，还有其他异步库可用于不同的场景，如 asyncpg 用于异步 PostgreSQL 数据库操作，aioredis 用于异步 Redis 操作。在编写异步代码时，要注意避免阻塞异步事件循环，确保所有 I/O 操作都是异步的。

示例

python

import asyncio
import aiohttp

async def fetch(session, url):
    # 异步发起 HTTP 请求
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        # 异步获取网页内容
        html = await fetch(session, 'http://example.com')
        print(html)

# 运行异步程序
asyncio.run(main())

7. 多线程共同操作同一个数据的互斥锁同步

多个线程同时操作同一个数据时，可能会导致数据不一致的问题，使用互斥锁可以解决这个问题。

要点

使用 threading.Lock() 创建互斥锁对象。
在访问共享数据前，使用 acquire 方法获取锁，确保只有一个线程能进入临界区。
操作完成后，使用 release 方法释放锁，允许其他线程访问。

python

import threading

# 共享数据
counter = 0
# 创建互斥锁
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        # 获取锁
        lock.acquire()
        try:
            counter += 1
        finally:
            # 释放锁
            lock.release()

threads = []
for _ in range(2):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(counter)

示例

在实际应用中，为了避免死锁，应尽量减少锁的持有时间，避免嵌套锁的使用。同时，可以使用 with 语句来简化锁的获取和释放操作。

python

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = []
for _ in range(2):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(counter)

8. 什么是多线程竞争

多线程竞争是指多个线程同时访问和修改共享资源时，由于执行顺序的不确定性，导致数据不一致或程序出现错误的现象。

要点

多个线程同时操作共享资源时，可能会出现数据竞争问题，如部分操作丢失。
线程执行顺序的不确定性是导致竞争的主要原因。
为了避免多线程竞争，可以使用同步机制，如互斥锁、信号量等。此外，还可以使用原子操作或无锁数据结构来提高并发性能。

示例

python

import threading

counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1

threads = []
for _ in range(2):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(counter)  # 结果可能不是 200000

9. 什么是线程同步

线程同步是多线程编程中确保数据一致性和程序正确性的重要手段，Python 提供了多种线程同步机制。

要点

互斥锁（Lock）确保同一时刻只有一个线程能访问共享资源。
递归锁（RLock）允许同一线程多次获取锁，避免嵌套调用时的死锁。
信号量（Semaphore）可以控制同时访问共享资源的线程数量。
条件变量（Condition）也是一种重要的线程同步机制，常用于生产者 - 消费者模型中。在使用线程同步机制时，要根据具体的场景选择合适的锁，避免死锁和性能问题。

示例

python

# 互斥锁示例
import threading

lock = threading.Lock()
shared_variable = 0

def increment_with_lock():
    global shared_variable
    for _ in range(100000):
        with lock:
            shared_variable += 1

threads = []
for _ in range(2):
    t = threading.Thread(target=increment_with_lock)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(shared_variable)

# 递归锁示例
import threading

rlock = threading.RLock()

def recursive_function():
    rlock.acquire()
    try:
        print("进入递归函数")
        if threading.current_thread().ident % 2 == 0:
            recursive_function()
    finally:
        rlock.release()

t = threading.Thread(target=recursive_function)
t.start()
t.join()

# 信号量示例
import threading
import time

semaphore = threading.Semaphore(2)

def worker():
    semaphore.acquire()
    try:
        print(f"{threading.current_thread().name} 开始工作")
        time.sleep(1)
        print(f"{threading.current_thread().name} 结束工作")
    finally:
        semaphore.release()

threads = []
for i in range(5):
    t = threading.Thread(target=worker, name=f"Thread-{i}")
    threads.append(t)
    t.start()

for t in threads:
    t.join()

10. 锁的概念与类型

锁是一种用于控制多个线程或进程对共享资源访问的机制，确保在同一时刻只有一个线程或进程能够访问共享资源，避免数据竞争和不一致。

互斥锁（Mutex）：Python 中的 threading.Lock 实现了互斥锁，同一时刻只有一个线程能获取锁。

python

import threading

lock = threading.Lock()
shared_data = 0

def access_shared_data():
    lock.acquire()
    try:
        global shared_data
        shared_data += 1
    finally:
        lock.release()

t = threading.Thread(target=access_shared_data)
t.start()
t.join()

递归锁（Reentrant Lock）：threading.RLock 是递归锁，允许同一线程多次获取锁，避免嵌套调用时的死锁。

python

import threading

rlock = threading.RLock()

def recursive_access():
    rlock.acquire()
    try:
        print("获取递归锁")
        if threading.current_thread().ident % 2 == 0:
            recursive_access()
    finally:
        rlock.release()

t = threading.Thread(target=recursive_access)
t.start()
t.join()

读写锁（Read - Write Lock）：允许多个线程同时进行读操作，但写操作会独占资源。可通过 threading.Lock 结合逻辑实现简单的读写锁。
信号量（Semaphore）：threading.Semaphore 可以控制同时访问共享资源的线程或进程数量，当信号量值为 1 时相当于互斥锁。

python

import threading
import time

semaphore = threading.Semaphore(2)

def worker():
    semaphore.acquire()
    try:
        print(f"{threading.current_thread().name} 开始工作")
        time.sleep(1)
        print(f"{threading.current_thread().name} 结束工作")
    finally:
        semaphore.release()

threads = []
for i in range(5):
    t = threading.Thread(target=worker, name=f"Thread-{i}")
    threads.append(t)
    t.start()

for t in threads:
    t.join()