闭包与回调函数：浅薄理解

工作之后，逐渐接触到了回调函数的应用。一开始我始终无法理解回调函数的执行顺序和意义，并且认为从可读性和执行效果上，并没有优于顺序执行的代码。不过后面对设计模式有一些了解后，对回调函数的概念大概有了一些模糊的理解，因此在这里整理一下以供学习。

回调函数

回调函数的主要特点是，将一个函数作为参数传入另外一个函数。并且在主函数执行完之后再继续调用该函数。

它和普通顺序写的函数相比有什么不同呢？在我的理解中，主要是能够有更好的扩展性。

例如以下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def data_processor(data):
    """
    处理数据的函数。直接在函数内部处理成功和失败的逻辑。
    """
    print(f"Processing data: {data}")
    # 假设根据数据的某些属性决定是否处理成功
    if len(data) > 5:  # 假设处理条件
        processed_data = data.upper()  # 模拟数据处理
        print(f"Data processed successfully: {processed_data}")
    else:
        error_message = "Data is too short."  # 错误信息
        print(f"Error processing data: {error_message}")

# 调用data_processor，直接处理数据
data_processor("Hello World")
data_processor("Hi")

上述代码在一个顺序执行的函数中判断是否处理数据。现在换成用回调函数来执行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def data_processor(data, success_callback, error_callback):
    """
    处理数据的函数。根据数据处理的结果，调用相应的回调函数。
    """
    print(f"Processing data: {data}")
    # 假设我们根据数据的某些属性决定是否处理成功
    if len(data) > 5:  # 假设条件
        processed_data = data.upper()  # 模拟数据处理
        success_callback(processed_data)  # 调用成功的回调函数
    else:
        error_message = "Data is too short."  # 错误信息
        error_callback(error_message)  # 调用错误的回调函数

def on_success(result):
    """
    数据处理成功时的回调函数。
    """
    print(f"Data processed successfully: {result}")

def on_error(error):
    """
    数据处理失败时的回调函数。
    """
    print(f"Error processing data: {error}")

# 调用data_processor，根据处理结果调用不同的回调函数
data_processor("Hello World", on_success, on_error)
data_processor("Hi", on_success, on_error)

这样有什么好处？

  1. 可以灵活的修改各自情况下的回调函数，不然就需要一直在主函数中修改。
  2. 所有逻辑包含在主函数中，并不利于维护。改成回调函数后，更利于调整逻辑。

异步操作

回调函数的另一个用法是在异步编程中使用，来避免操作阻塞程序。这种用法是在一个操作完成时（例如，网络请求、读取文件等），执行后续的操作。

以下是示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import asyncio
import aiohttp

async def fetch_data(url):
    """
    异步获取URL的数据
    """
    print(f"Starting to fetch data from {url}")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            return data

async def on_data_fetched(data):
    """
    模拟回调函数：数据获取后被调用
    """
    print("Data fetched: ")
    print(data[:100])  # 假设只打印数据的前100个字符

async def main():
    """
    主函数，执行异步的数据获取操作，并在完成后处理数据
    """
    url = "https://www.example.com"
    data = await fetch_data(url)
    await on_data_fetched(data)

# 运行主函数
asyncio.run(main())

上述代码中，on_data_fetched函数在fetch_data函数完成后被调用。它的好处是可以在等待执行的时候运行其他的程序部分。

闭包

相比于回调函数，闭包这个概念更多的涉及到作用域。它允许函数访问并操作函数外部的变量。我的理解是，它的作用是创建一个定义作用域，这个作用域能够获取外部的变量。

以下是一个示例：

1
2
3
4
5
6
7
8
def foo():
	a = 'free var'
	def bar():
	    print(a)
	return bar

foo()()
# free var

可以看到，bar函数能够读取到外部的a变量，这就是闭包。

它和之前看到的装饰器有什么区别？我在查阅了一些资料后得到以下的结论：

装饰器是闭包思想的一种实现。而闭包的定义是在函数内定义另一个函数。内函数能够调用外函数的临时变量。

例如给一个示例闭包函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def outfun(outNum):  
    print('---start outfun---')  

    def infun(inNum):             
        print('---start infun---')  
        print(outNum + inNum)  
        print('---end infun---')  

    print('---end outfun---')  
    return infun  

ret = outfun(10)  
ret(20) 
# ---start outfun---  
# ---end outfun---  
# ---start infun---  
# 30  
# ---end infun---

上面的ret本身是一个函数，它指向infun()，并且将其中的outNum定义为10。之后在调用ret时输入的参数则充当了infun的参数。

我们再来示范一下装饰器的用法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def test_func1():
    print('Begin func1...')

def test_func2():
    print('Begin func2...')
    
def check(function):
    def check_inner():
        print('Checking Inner Function...')
        function()
  	return check_inner

check_test1 = check(test_func1)
check_test1()

# 也可以用语法糖实现
@check
def test_func1():
    print('Begin func1...')

@check
def test_func2():
    print('Begin func2...')

test_func1()
test_func2()

在check函数中，我们设定了一个内函数，它接收外部的函数作为参数，指向内部的函数。在内部函数中对外部函数进行了一系列处理。这就是装饰器的简单用法。

更常用的用法是不定长参数的使用，例如：

1
2
3
4
def deco(func):
    def new_func(*args, **kwargs):
        return func(*args, **kwargs)
    return new_func

以及接收参数的装饰器工厂：

1
2
3
4
5
6
7
8
9
def deco_factory(*args, **kwargs):
    def deco(func):
        print(args)
        return func
    return deco

@deco_factory('factory')
def foo():
    pass

这样就能灵活修改装饰器的方法。

最后，记录一个接收参数的装饰器用法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import time

def outer(choose): # 在最外层函数中加入参数
  if choose==1: # 通过choose参数，选择装饰器
    def deco(func):
      def inner(*arg, **kwarg):
        print('decoration1')
        begin_time = time.time()
        time.sleep(2) # 睡眠2s
        a = func(*arg, **kwarg) 
        end_time = time.time()
        print('运行时间1：', end_time - begin_time)
        return a
      return inner
    return deco
  
  else:
    def deco(func):
      def inner(*arg, **kwarg): 
        print('decoration2')
        begin_time = time.time()
        time.sleep(5) # 睡眠5s
        a = func(*arg, **kwarg) 
        end_time = time.time()
        print('运行时间2：', end_time - begin_time)
        return a
      return inner
    return deco

@outer(1) # 由于outer中有参数，此处必须传入参数
def test1(a):
  print('test function1:', a)
  return a

@outer(5) # 传入另一个参数
def test2(a):
  print('test function2:', a)
  return a


# 分别调用2个函数（2个函数装饰器相同，装饰器参数不同）
test1(2) # 调用test1
  
decoration1
test function1: 2
运行时间1： 2.000072717666626 # 2秒
 # test1的返回值

test2(4) # 调用test2
  
decoration2
test function2: 4
运行时间2： 5.000797986984253 # 5秒
 # test2的返回值

2024/4/3 于苏州