許多函數式文章講述的是組合、流水線和高階函數這樣的抽象函數式技術。本文不同，它展示了人們每天編寫的命令式、非函數式代碼示例，以及將這些示例轉換為函數式風格。

文章的第一部分將一些短小的數據轉換循環重寫成函數式的 maps 和 reduces 。第二部分選取長一點的循環，把他們分解成單元，然后把每個單元改成函數式的。第三部分選取一個很長的連續數據轉換循環，然后把它分解成函數式流水線。

示例都是用 Python 寫的，因為很多人覺得 Python 易讀。為了證明函數式技術對許多語言來說都相同，許多示例避免使用 Python 特有的語法：map、reduce、pipeline 。

導引

當人們談論函數式編程，他們會提到非常多的 “函數式” 特性。提到不可變數據¹、一等函數²以及尾調用優化³，這些是幫助函數式編程的語言特征。提到 mapping（映射）、reducing（歸納）、pipelining（管道）、recursing（遞歸）、currying⁴（科里化）以及高階函數的使用，這些是用來寫函數式代碼的編程技術。提到并行⁵、惰性計算⁶以及確定性⁷，這些是有利于函數式編程的屬性。

忽略全部這些，可以用一句話來描述函數式代碼的特征：避免副作用。它不會依賴、也不會改變當前函數以外的數據。所有其他的 “函數式” 的東西都源于此。當你學習本文時請將這句話作為指引。

這是一個非函數式方法：

a = 0 def increment():global aa += 1

這是一個函數式的方法：

def increment(a):return a + 1

不要在lists上迭代。使用map和reduce。

Map（映射）

Map 接受一個方法和一個集合作為參數。它創建一個新的空集合，以集合中每個元素作為參數，調用傳入的方法，然后將返回值插入到新創建的集合中。最后返回這個新集合。

這是一個簡單的 map，接受一個存放名字的 list，返回一個存放名字長度的 list：

name_lengths = map(len, ["Mary", "Isla", "Sam"])print(list(name_lengths)) # => [4, 4, 3]

接下來這個 map 將傳入的 collection 中每個元素都做平方操作：

squares = map(lambda x: x * x, [0, 1, 2, 3, 4])print(list(squares)) # => [0, 1, 4, 9, 16]

這個 map 并沒有使用一個命名的方法。它是使用了一個用 lambda定義的，匿名并且內聯的方法。lambda 的參數定義在冒號左邊，方法主體定義在冒號右邊，返回值是方法體運行的結果。

下面的非函數式代碼接受一個真名列表，然后用隨機指定的代號來替換真名。

import randomnames = ['Mary', 'Isla', 'Sam'] code_names = ['Mr. Pink', 'Mr. Orange', 'Mr. Blonde']for i in range(len(names)):names[i] = random.choice(code_names)print(names) # => ['Mr. Blonde', 'Mr. Blonde', 'Mr. Blonde']

（正如你所見的，這個算法可能會給多個密探同一個秘密代號。希望這不會在任務中混淆。）

現在可以用 map 重寫：

import randomnames = ['Mary', 'Isla', 'Sam']secret_names = map(lambda x: random.choice(['Mr. Pink','Mr. Orange','Mr. Blonde']),names)

練習1：嘗試用 map 重寫下面的代碼。它接受由真名組成的 list 作為參數，然后用一個更加穩定的策略產生一個代號來替換這些名字。

names = ['Mary', 'Isla', 'Sam']for i in range(len(names)):names[i] = hash(names[i])print(names) # => [-3894872169223495002, -1027760254031825677, -767524867397611892]

（希望密探記憶力夠好，不要在執行任務時把代號忘記了。）

我的解決方案：

names = ['Mary', 'Isla', 'Sam']secret_names = map(hash, names)

Reduce（迭代）

Reduce 接受一個方法和一個集合做參數。返回通過這個方法迭代容器中所有元素產生的結果。

這是個簡單的 reduce。返回集合中所有元素的和。

from functools import reducesum = reduce(lambda a, x: a + x, [0, 1, 2, 3, 4])print(sum) # => 10

x是迭代的當前元素。a是累加和也就是在之前的元素上執行lambda返回的值。reduce()遍歷元素。每次迭代，在當前的 a和 x上執行 lambda 然后返回結果作為下一次迭代的 a。

第一次迭代的 a是什么？在這之前沒有迭代結果傳進來。reduce()使用集合中的第一個元素作為第一次迭代的 a，然后從第二個元素開始迭代。也就是說，第一個 x是第二個元素。

下面這段代碼記 'Sam'這個單詞在字符串列表中出現的頻率：

sentences = ['Mary read a story to Sam and Isla.','Isla cuddled Sam.','Sam chortled.']sam_count = 0 for sentence in sentences:sam_count += sentence.count('Sam')print(sam_count) # => 3

下面是用 reduce 寫的：

from functools import reducesentences = ['Mary read a story to Sam and Isla.','Isla cuddled Sam.','Sam chortled.']sam_count = reduce(lambda a, x: a + x.count('Sam'),sentences,0)

這段代碼如何初始化 a？出現 'Sam'的起始點不能是 ’Mary read a story to Sam and Isla.’。初始的累加和 a 由 reduce()的第三個參數來指定。這樣就允許了集合中元素的類型可以與累加器不同。

為什么 map 和 reduce 更好？

首先，它們大多是一行代碼。

其次，迭代中最重要的部分：集合、操作和返回值，在所有的 map 和 reduce 中總是在相同的位置。

再次，循環（for）中的代碼可能會改變之前定義的變量或之后要用到的變量。照例，maps 和 reduces 是函數式的。

然后，map 和 reduce 是元素操作。每次有人讀到 for循環，他們都要逐行讀懂邏輯。幾乎沒有什么規律性的結構可以幫助理解代碼。相反，map 和 reduce 都是創建代碼塊來組織復雜的算法，并且讀者也能非常快的理解元素，并在腦海中抽象出來?！鞍?#xff0c;這段代碼正在轉換集合中的每一個元素。它丟棄了一些轉換結果。它把剩下的組合成一個輸出。”

最后，map 和 reduce 有許多提供便利的 “好朋友”，它們是基本行為的修訂版。例如 filter、all、any以及 find。

練習2：嘗試用 map 、reduce 和 filter 重寫下面的代碼。Filter 接受一個方法和一個集合，返回集合中使方法返回 True的元素。

people = [{'name': 'Mary', 'height': 160},{'name': 'Isla', 'height': 80},{'name': 'Sam'}]height_total = 0 height_count = 0 for person in people:if 'height' in person:height_total += person['height']height_count += 1if height_count > 0:average_height = height_total / height_countprint(average_height)# => 120.0

如果這個比較棘手，試著不要考慮數據上的操作?？紤]下數據要經過的狀態，從 people 字典列表到平均高度。不要嘗試把多個轉換捆綁在一起。把每一個放在獨立的一行，并且把結果保存在命名良好的變量中。代碼可以運行后，再提煉。

我的方案：

from functools import reducepeople = [{'name': 'Mary', 'height': 160},{'name': 'Isla', 'height': 80},{'name': 'Sam'}]heights = list(map(lambda x: x['height'],filter(lambda x: 'height' in x, people)))if len(heights) > 0:from operator import addaverage_height = reduce(add, heights) / len(heights)

寫聲明式代碼，而不是命令式

下面的程序演示三輛車比賽。每次可移動時間，每輛車可能移動或者不動。每次可移動時間，程序會打印到目前為止所有車的路徑。五次后，比賽結束。

下面是某一次的輸出：

- -- ---- --- ---- ---- ---- ----- ------ ------ ----

這是程序：

from random import randomtime = 5 car_positions = [1, 1, 1]while time:# decrease timetime -= 1print()for i in range(len(car_positions)):# move carif random() > 0.3:car_positions[i] += 1# draw carprint('-' * car_positions[i])

代碼是命令式的。一個函數式的版本應該是聲明式的。應該描述要做什么，而不是怎么做。

使用函數

通過綁定代碼片段到函數里，可以使程序更有聲明式的味道。

from random import randomdef move_cars():for i, _ in enumerate(car_positions):if random() > 0.3:car_positions[i] += 1def draw_car(car_position):print('-' * car_position)def run_step_of_race():global timetime -= 1move_cars()def draw():print()for car_position in car_positions:draw_car(car_position)time = 5 car_positions = [1, 1, 1]while time:run_step_of_race()draw()

想要理解這段代碼，讀者只需要看主循環：“如果 time 不為 0，運行一下 run_step_of_race 和 draw ，再檢查一下 time 。”如果讀者想更多地理解這段代碼中的 run_step_of_race 或 draw ，可以自行閱讀函數對應的代碼。

代碼里沒有注釋，因為這段代碼是自描述的。

把代碼分解成函數是非常好的，這樣可以提高代碼的可讀性。

上面的代碼用到了函數，但也只是將函數作為子程序來使用，最后把它們打包成代碼。根據本文導引的定義，上述代碼不是函數式的。因為代碼中的函數使用了狀態，并且這個狀態不是通過函數參數傳入的。函數通過改變外部變量，而不是通過返回值，來影響了其他代碼的行為。為了搞清楚函數的真正行為，讀者必須仔細閱讀每行代碼。如果發現一個外部變量，你就必須先找到它定義的地方，再尋找有哪些函數會修改這個變量。

移除狀態

下面是函數式的版本：

from random import randomdef move_cars(car_positions):return list(map(lambda x: x + 1 if random() > 0.3 else x,car_positions))def output_car(car_position):return '-' * car_positiondef run_step_of_race(state):return {'time': state['time'] - 1,'car_positions': move_cars(state['car_positions'])}def draw(state):print()print('\n'.join(map(output_car, state['car_positions'])))def race(state):draw(state)if state['time']:race(run_step_of_race(state))race({'time': 5,'car_positions': [1, 1, 1]})

代碼仍然是分解成不同的函數，但是這段代碼是函數式的。

函數式的函數有三個特征：

沒有共享變量。

time和 car_positions直接傳進方法 race()中。

函數接受參數。

函數里沒有實例化變量。

所有的數據變化都通過返回值實現。race()使用 run_step_of_race()的結果進行遞歸³。每次遞歸的 step 都產生一個新狀態，這個狀態會直接傳遞給下一個 step 。

現在，有兩個函數，zero()和 one()：

def zero(s):if s[0] == "0":return s[1:]def one(s):if s[0] == "1":return s[1:]

zero()接受一個字符串 s作為參數，如果第一個字符是 '0'，方法返回字符串的其他部分。如果不是，返回 None，Python 的默認返回值。one()做的事情相同，除了第一個字符要求是 '1'。

想象一個名為 rule_sequence()的函數，接受一個 string 和一個 list，list 用來存放由 zero()和 one()組成的一系列規則函數。在 string 上調用第一個規則。除非返回 None，不然它會繼續接受返回值并且在 string 上調用第二個規則。除非返回 None，不然它會接受返回值，并且調用第三個規則。等等。如果有哪一個規則返回 None，rule_sequence()方法停止，并返回 None。不然，返回最后一個規則方法的返回值。

下面是一個示例：

print(rule_sequence('0101', [zero, one, zero])) # => 1print(rule_sequence('0101', [zero, zero])) # => None

這是 rule_sequence()的命令式版本：

def rule_sequence(s, rules):for rule in rules:s = rule(s)if s == None:breakreturn s

練習3：上面的代碼用循環來完成功能。用遞歸重寫使它更有聲明式的味道。

我的方案：

def rule_sequence(s, rules):if s == None or not rules:return selse:return rule_sequence(rules[0](s), rules[1:])

使用流水線

前上一節中，一些命令式的循環被重寫為調用輔助函數的遞歸形式。在本節中，會用一種稱為 pipeline 的技術重寫為另一種形式的命令式循環。

下面的 list 存放了三個字典型的數據，每個字典存放一個樂隊相關的三個鍵值對：姓名、不正確的國籍和激活狀態。format_bands()函數循環處理這個 list 。

bands = [{'name': 'sunset rubdown', 'country': 'UK', 'active': False},{'name': 'women', 'country': 'Germany', 'active': False},{'name': 'a silver mt. zion', 'country': 'Spain', 'active': True }]def format_bands(bands):for band in bands:band['country'] = 'Canada'band['name'] = band['name'].replace('.', '')band['name'] = band['name'].title()format_bands(bands)print(bands) # => [{'name': 'Sunset Rubdown', 'country': 'Canada', 'active': False}, {'name': 'Women', 'country': 'Canada', 'active': False}, {'name': 'A Silver Mt Zion', 'country': 'Canada', 'active': True}]

函數名會引起一些擔憂?！眆ormat” 是一個很模糊的詞。仔細檢查代碼以后，這些擔憂開始顯現。循環中做了三件事：鍵為 'country'的值被設置為 'Canada'；姓名中的標點符號被移除了；姓名首字母改成了大寫。但是很難看出這段代碼的目的是什么，是否做了它看上會去做的事。并且代碼難以重用、難以測試已經難以并行。

和下面這段代碼比較一下：

print(pipeline_each(bands, [set_canada_as_country,strip_punctuation_from_name,capitalize_names]))

這段代碼很容易理解。它去除了副作用，輔助函數給人的感覺是函數式的，因為它們看上去被鏈接在一起。前一個函數的輸出構成下一個函數的輸入。如果這些方法是函數式的，那么就很容易進行驗證。它們很容易重用、測試并且也很容易并行。

pipeline_each()的工作是傳遞 bands，一次傳遞一個元素給轉換函數，比如 set_cannada_as_country()。當 bands 中的所有元素都用來調用過這個函數之后，pipeline_each()將轉換后的 bands 收集起來。再依次傳遞給下一個函數。

我們來看看轉換函數。

def assoc(_d, key, value):from copy import deepcopyd = deepcopy(_d)d[key] = valuereturn ddef set_canada_as_country(band):return assoc(band, 'country', "Canada")def strip_punctuation_from_name(band):return assoc(band, 'name', band['name'].replace('.', ''))def capitalize_names(band):return assoc(band, 'name', band['name'].title())

每一個函數都將 band 的一個 key 關聯到一個新的 value 上。這在不改變原始 band 的情況下是很難做到的。assoc()通過使用 deepcopy()生成傳入的字典的一個拷貝來解決這個問題。每個轉換函數修改這個拷貝，然后將這個拷貝返回。

一切看上去好像都很完美。原始的 band 字典不再擔心因為某個 key 需要關聯新的 value 而被改變。但是上面的代碼有兩個潛在的副作用。在 strip_punctuation_from_name()中，不含標點的姓名是通過在原值上調用 replace()方法產生的。在 capitalize_names()中，姓名的首字母大寫是通過在原值上調用 title()產生的。如果 replace()和 title()不是函數式的，那么 strip_punctuation_from_name()和 capitalize_names()也就不是函數式的。

幸運的是，replace()和 title()并不改變它們所操作的字符串。因為 Python 中的字符串是不可變的。當 replace()操作 band 的姓名字符串時，也是先拷貝原始字符串，然后再對拷貝的字符串做修改。嘖嘖。

Python 中字符串和字典之間可變性的不同，突顯出類似 Clojure 這類語言的吸引力。程序員永遠不用擔心數據是否可變，數據是不可變的。

練習4：試著寫出 pipeline_each函數?？紤]操作的順序，首先傳遞數組中的每個 band 給第一個轉換函數，每次傳遞一個 ；接著傳遞結果數組中的每個 band 給第二個轉換函數，每次傳遞一個。等等。

我的方案：

def pipeline_each(data, fns):return reduce(lambda a, x: map(x, a),fns,data)

歸根到底，三個轉換函數都是對傳入的 band 的一個特定字段進行更改。call()可以用來抽象這個功能。call()接受一個函數和一個鍵作為參數。

set_canada_as_country = call(lambda x: 'Canada', 'country') strip_punctuation_from_name = call(lambda x: x.replace('.', ''), 'name') capitalize_names = call(str.title, 'name')print pipeline_each(bands, [set_canada_as_country,strip_punctuation_from_name,capitalize_names])

或者，如果我們希望能滿足簡潔方面的可讀性，那么就：

print pipeline_each(bands, [call(lambda x: 'Canada', 'country'),call(lambda x: x.replace('.', ''), 'name'),call(str.title, 'name')])

call()的代碼：

def assoc(_d, key, value):from copy import deepcopyd = deepcopy(_d)d[key] = valuereturn ddef call(fn, key):def apply_fn(record):return assoc(record, key, fn(record.get(key)))return apply_fn

這段代碼做了很多事情，讓我們一點一點的看。

• 首先，call()是一個高階函數。高階函數接受一個函數作為參數，或者返回一個函數。或者像 call()，兩者都有。

• 其次，apply_fn()看起來很像那三個轉換函數。它接受一個 record（一個 band），查找在 record[key]的值，以這個值為參數調用 fn，指定 fn的結果返回到 record 的拷貝中，然后返回這個拷貝。

• 再次，call()沒有做任何實際的工作。它被調用時，實際上是apply_fn()真正在工作。上面使用 pipeline_each()的例子中，一個 apply_fn()的實例將傳入的 band 的 'country'設為 'Canada'。另一個實例將傳入的 band 中的名字的首字母設為大寫。

• 然后，當一個 apply_fn()實例運行時，fn和 key將不在作用域中。它們既不是 apply_fn()的參數，也不是其中的局部變量。但是它們仍然可以被訪問。當一個函數被定義時，它會保存方函數結束前包含的變量的引用：那些定義在函數的作用域外，卻在函數中使用的變量。當函數運行并且代碼引用一個變量時，Python 會查找局部和參數中的變量。如果沒找到，就會去找閉包內保存的變量。那就是找到 fn和 key的地方。

• 最后，在 call()的代碼中沒有提到 bands。因為不管程序的主題是什么，call()都可以為其生成流水線函數。函數式編程的一部分目的就是構建一個通用、可重用、可組合的函數庫。

干的漂亮。閉包、高階函數和變量作用域都被包含在本節中。喝杯可口的檸檬水吧。

還需要在 band 上做一點處理。就是移除 name 和 country 以外的所有東西。extract_name_and_country()可以實現這個目標。

def extract_name_and_country(band):plucked_band = {}plucked_band['name'] = band['name']plucked_band['country'] = band['country']return plucked_bandprint pipeline_each(bands, [call(lambda x: 'Canada', 'country'),call(lambda x: x.replace('.', ''), 'name'),call(str.title, 'name'),extract_name_and_country])# => [{'name': 'Sunset Rubdown', 'country': 'Canada'}, # {'name': 'Women', 'country': 'Canada'}, # {'name': 'A Silver Mt Zion', 'country': 'Canada'}]

extract_name_and_country()可以寫成名為 pluck()的通用函數。pluck()可以這樣使用：

print pipeline_each(bands, [call(lambda x: 'Canada', 'country'),call(lambda x: x.replace('.', ''), 'name'),call(str.title, 'name'),pluck(['name', 'country'])])

練習5：pluck()接受一系列的鍵值，根據這些鍵值從 record 中提取數據。試著寫寫。需要用到高階函數。

我的方案：

def pluck(keys):def pluck_fn(record):return reduce(lambda a, x: assoc(a, x, record[x]),keys,{})return pluck_fn

還有什么要做的嗎？

函數式代碼可以很好的和其他風格的代碼配合使用。文章中的轉換器可以用在任何代碼里面，并且可以用任何語言實現。試試將它們用在你自己的代碼里。

想想 Mary、Isla 和 Sam 。是如何將對 list 的迭代操作，轉換成 map 和 reduce 操作？

想想汽車賽。是如何將代碼分解成函數，再將函數改造成函數式，最后把循環的重復處理轉換成遞歸操作的？

想想樂隊。是如何將一系列的操作轉換成流水線（pipeline）的？

標注

英文原文：A practical introduction to functional programming

譯文原文：函數式編程實戰教程（Python版） —— 鏈接已失效

---

一塊不可變數據是指不能被改變的數據。一些語言（比如 Clojure），默認所有的值都是不可變的。任何的可變操作都是通過拷貝值，并返回修改后的拷貝來實現的。這樣就消除了程序中訪問未完成狀態，所造成的 bug。 ??

支持一等函數的語言允許像處理其他類型的值那樣處理函數。意味著函數可以被創建、傳給其他函數、從函數中返回以及存儲在其他數據結構里。 ??

尾調用優化是一種編程語言特性。每次函數遞歸，都會創建一個棧。棧用來存儲當前函數需要使用的參數和局部變量。如果一個函數遞歸次數非常多，很可能會讓編譯器或解釋器消耗掉所有的內存。有尾調用優化的語言會通過重用同一個棧來支持整個遞歸調用的序列。像 Python 這樣不支持尾調用優化的語言，通常會限制函數遞歸的數量在千次級別。在 race()方法中，只有5次，所以很安全。 ?? ??

Currying 的意思是：將接受多個參數的函數分解成一個只接受第一個參數的函數，這個函數返回一個接受第二個參數的函數，以此類推，直到接受完所有參數。 ??

并行的意思是在不同步的情況下并發地運行同一段代碼。這些并發操作常常運行在不同的處理器上。 ??

惰性計算是編譯器的技術，用來避免在真正需要代碼的結果之前，就運行了代碼。 ??

只有當每次重復都能得出相同的結果，才能說處理是確定性的。 ??

總結

以上是生活随笔為你收集整理的函数式编程实战教程（Python版）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。