python并发3：使用asyncio编写服务器
asyncio 上一篇我们介绍了 asyncio 包，以及如何使用异步编程管理网络应用中的高并发。在这一篇，我们主要介绍使用 asyncio 包编程的两个例子。 async/await语法 我们先介绍下 async/await 语法，要不然看完这篇可能会困惑，为什么之前使用 asyncio.coroutine 装饰器 和 yield from，这里都是 用的 async 和 await？ python并发2：使用asyncio处理并发 async/await 是Python3.5 的新语法，语法如下： async def read_data(db): pass async 是明确将函数声明为协程的关键字，即使没有await表达式，函数执行也会返回一个协程对象。 在协程函数内部，可以在某个表达式之前使用 await 关键字来暂停协程的执行，以等待某协程完成： async def read_data(db): data = await db.fetch('SELECT ...') 这个代码如果使用 asyncio.coroutine 装饰器语法为： @asyncio.coroutine def read_data(db): data = yield from db.fetch('SELECT ...') 这两段代码执行的结果是一样的，也就是说 可以把 asyncio.coroutine 替换为 async， yield from 替换为 await。 使用新的语法有什么好处呢： 使生成器和协程的概念更容易理解，因为语法不同 可以消除由于重构时不小心移出协程中yield 声明而导致的不明确错误，这回导致协程变成普通的生成器。 使用 asyncio 包编写服务器 这个例子主要是使用 asyncio 包 和 unicodedata 模块，实现通过规范名称查找Unicode 字符。 我们先来看一下代码： # charfinder.py import sys import re import unicodedata import pickle import warnings import itertools import functools from collections import namedtuple RE_WORD = re.compile('\w+') RE_UNICODE_NAME = re.compile('^[A-Z0-9 -]+$') RE_CODEPOINT = re.compile('U\+[0-9A-F]{4, 6}') INDEX_NAME = 'charfinder_index.pickle' MINIMUM_SAVE_LEN = 10000 CJK_UNI_PREFIX = 'CJK UNIFIED IDEOGRAPH' CJK_CMP_PREFIX = 'CJK COMPATIBILITY IDEOGRAPH' sample_chars = [ '$', # DOLLAR SIGN 'A', # LATIN CAPITAL LETTER A 'a', # LATIN SMALL LETTER A '\u20a0', # EURO-CURRENCY SIGN '\u20ac', # EURO SIGN ] CharDescription = namedtuple('CharDescription', 'code_str char name') QueryResult = namedtuple('QueryResult', 'count items') def tokenize(text): ''' :param text: :return: return iterable of uppercased words ''' for match in RE_WORD.finditer(text): yield match.group().upper() def query_type(text): text_upper = text.upper() if 'U+' in text_upper: return 'CODEPOINT' elif RE_UNICODE_NAME.match(text_upper): return 'NAME' else: return 'CHARACTERS' class UnicodeNameIndex: # unicode name 索引类 def __init__(self, chars=None): self.load(chars) def load(self, chars=None): # 加载 unicode name self.index = None if chars is None: try: with open(INDEX_NAME, 'rb') as fp: self.index = pickle.load(fp) except OSError: pass if self.index is None: self.build_index(chars) if len(self.index) > MINIMUM_SAVE_LEN: try: self.save() except OSError as exc: warnings.warn('Could not save {!r}: {}' .format(INDEX_NAME, exc)) def save(self): with open(INDEX_NAME, 'wb') as fp: pickle.dump(self.index, fp) def build_index(self, chars=None): if chars is None: chars = (chr(i) for i in range(32, sys.maxunicode)) index = {} for char in chars: try: name = unicodedata.name(char) except ValueError: continue if name.startswith(CJK_UNI_PREFIX): name = CJK_UNI_PREFIX elif name.startswith(CJK_CMP_PREFIX): name = CJK_CMP_PREFIX for word in tokenize(name): index.setdefault(word, set()).add(char) self.index = index def word_rank(self, top=None): # (len(self.index[key], key) 是一个生成器，需要用list 转成列表，要不然下边排序会报错 res = [list((len(self.index[key], key)) for key in self.index)] res.sort(key=lambda item: (-item[0], item[1])) if top is not None: res = res[:top] return res def word_report(self, top=None): for postings, key in self.word_rank(top): print('{:5} {}'.format(postings, key)) def find_chars(self, query, start=0, stop=None): stop = sys.maxsize if stop is None else stop result_sets = [] for word in tokenize(query): # tokenize 是query 的生成器 a b 会是 ['a', 'b'] 的生成器 chars = self.index.get(word) if chars is None: result_sets = [] break result_sets.append(chars) if not result_sets: return QueryResult(0, ()) result = functools.reduce(set.intersection, result_sets) result = sorted(result) # must sort to support start, stop result_iter = itertools.islice(result, start, stop) return QueryResult(len(result), (char for char in result_iter)) def describe(self, char): code_str = 'U+{:04X}'.format(ord(char)) name = unicodedata.name(char) return CharDescription(code_str, char, name) def find_descriptions(self, query, start=0, stop=None): for char in self.find_chars(query, start, stop).items: yield self.describe(char) def get_descriptions(self, chars): for char in chars: yield self.describe(char) def describe_str(self, char): return '{:7}\t{}\t{}'.format(*self.describe(char)) def find_description_strs(self, query, start=0, stop=None): for char in self.find_chars(query, start, stop).items: yield self.describe_str(char) @staticmethod # not an instance method due to concurrency def status(query, counter): if counter == 0: msg = 'No match' elif counter == 1: msg = '1 match' else: msg = '{} matches'.format(counter) return '{} for {!r}'.format(msg, query) def main(*args): index = UnicodeNameIndex() query = ' '.join(args) n = 0 for n, line in enumerate(index.find_description_strs(query), 1): print(line) print('({})'.format(index.status(query, n))) if __name__ == '__main__': if len(sys.argv) > 1: main(*sys.argv[1:]) else: print('Usage: {} word1 [word2]...'.format(sys.argv[0])) 这个模块读取Python内建的Unicode数据库，为每个字符名称中的每个单词建立索引，然后倒排索引，存入一个字典。 例如，在倒排索引中，‘SUN’ 键对应的条目是一个集合，里面是名称中包含’SUN’ 这个词的10个Unicode字符。倒排索引保存在本地一个名为charfinder_index.pickle 的文件中。如果查询多个单词，会计算从索引中所得集合的交集。 运行示例如下： ...