Python常用编程模块汇总

一、关于文件处理

1、json文件读写

2、xml文件读写

3、csv文件读写

4、excel文件读写

二、关于异步

1、threading

1、asyncio

2、aio_pika库

一、关于文件处理

1、json文件读写

参考：JSON 编码和解码器

import json
f_name = 'msginp_10_11.json'
msg_inp = input('请输入您喜欢的数字：')# json.dumps()和json.loads()是json格式处理函数
# json.dumps()函数将字典转化为字符串
# json.loads()函数将字符串转化为字典# write_json数据
with open(f_name, 'w') as fn:json.dump(msg_inp, fn)   #实际作用等同于如下两行# json_str = json.dumps(msg_inp)   #将字符串转换为json格式# fn.write(json_str)   #将json格式的字符串写入文件#read json数据
with open(f_name) as fn1:c = json.load(fn1)print('I know your favorite number! It\'s '+ c)

2、xml文件读写

参考：XML处理模块 — Python 3.9.1 文档

基于ElementTree实现xml文件读写，xml示例如下：

<?xml version="1.0"?>
<data><country name="Liechtenstein"><rank>1</rank><year>2008</year><gdppc>141100</gdppc><neighbor name="Austria" direction="E"/><neighbor name="Switzerland" direction="W"/></country><country name="Singapore"><rank>4</rank><year>2011</year><gdppc>59900</gdppc><neighbor name="Malaysia" direction="N"/></country><country name="Panama"><rank>68</rank><year>2011</year><gdppc>13600</gdppc><neighbor name="Costa Rica" direction="W"/><neighbor name="Colombia" direction="E"/></country>
</data>

常见操作如下：

import xml.etree.ElementTree as ET# 1.获取xml树有两种方式：读取xml文件获取或者读取xml格式的字符串获取
# 1）读取xml文件获取
tree = ET.parse('country_data.xml')
root = tree.getroot()# 2）读取xml格式的字符串获取：返回的是根节点
root = ET.fromstring(country_data_as_string)# 2.节点的标签和属性
root.tag
root.attrib
root[0][1].text# 3.迭代节点
for child in root:# 4.通过索引访问特定的子级节点：
root[0][1].text# 5.查找节点
for neighbor in root.iter('neighbor'):
for country in root.findall('country'):# 6.修改xml文件
Element.text #修改内容的值
Element.set() #添加或修改属性值
Element.append() #添加节点
a = ET.Element('a')
b = ET.SubElement(a, 'b') #向节点a添加子节点b
Element.remove() #删除节点

3、csv文件读写

4、excel文件读写

二、关于异步

1、threading

参考：python多线程详解（超详细）

import threading
from threading import Lock,Thread
import time
import os# 多个线程共享全局变量
g_num = 100def run(n):print('task',n)time.sleep(1)print('2s')time.sleep(1)print('1s')time.sleep(1)print('0s')time.sleep(1)# 1.创建线程：# target是要执行的函数名（不是函数），args是函数对应的参数，以元组的形式存在
t1 = threading.Thread(target=run, args=('t1',))
t2 = threading.Thread(target=run, args=('t2',))# 2.启动线程
t1.start()
t2.start()# 3.声明线程锁
lock = Lock()
lock.acquire()
lock.release()# 4.声明信号量
semaphore = threading.BoundedSemaphore(5)   #最多允许5个线程同时运行
semaphore.acquire()   #加锁
semaphore.release()    #释放# 5.事件：全局定义了一个Flag，当Flag的值为False，那么event.wait()就会阻塞，当flag值为True，那么event.wait()便不再阻塞
event = threading.Event()
event.set()
event.clear()
event.wait()

1、asyncio

参考：协程与任务

import asyncio
import time# 1.定义coroutine，使用async/await
async def test(delay, what):await asyncio.sleep(delay)print(what)# 2.获得task
task_list = []
for i in range(10):task = asyncio.ensure_future(test(delay=1, what=”hello“))task_list(task)# 3.获取Future对象
tasks = asyncio.gather(* task_list)if __name__ == "__main__":loop = asyncio.get_event_loop()loop.run_until_complete(tasks)loop.close()

python并发大量协程时报错ValueError: too many file descriptors in select()。这个是操作系统关于select()的限制，linux打开文件的最大数默认是1024，windows默认是509，超过了这个值，程序就开始报错。这里我们有三种方法解决这个问题：

限制并发数量。（一次不要塞那么多任务，或者限制最大并发数量）
使用回调的方式。
修改操作系统打开文件数的最大限制，在系统里有个配置文件可以修改默认值，具体步骤不再说明了。

2、aio_pika库

参考：aio-pika中文文档

import asyncio
import aio_pika# 1.建立连接
async def get_connection():connection_string = "amqp://{user}:{password}@{mq_ip}:{mq_port}".format(user="user", password="password",mq_ip="mq_ip", mq_port="5672")connection = await aio_pika.connect_robust(connection_string, loop=loop, ssl=True)return connection# 2.向交换机推送消息
async def publish(routing_key="test_exchange", exchange_name="test_exchange"):connection = await get_connection()async with connection:channel = await connection.channel()exchange = await channel.get_exchange(exchange_name)message = aio_pika.Message(body="Hello {}".format(routing_key).encode())await exchange.publish(message, routing_key=routing_key)# 3.从队列中获取消息
async def consume(queue_name="test_queue"):connection = await get_connection()async with connection:channel = await connection.channel()queue = await channel.get_queue(queue_name)async with queue.iterator() as queue_iter:async for message in queue_iter:async with message.process():print(message.body)if queue.name in message.body.decode():breaktask_list = []
for i in range(2):task = asyncio.ensure_future(publish(routing_key="test_exchange", exchange_name="test_exchange"))task_list.append(task)
tasks = asyncio.gather(*task_list)if __name__ == "__main__":loop = asyncio.get_event_loop()loop.run_until_complete(tasks)loop.close()

连接报错：Rabbitmq报错pika.exceptions.IncompatibleProtocolError: StreamLostError: ('Transport indicated EOF',)

报错原因：产生此报错的原因是将rabbit连接port写成了15672。rabbitmq需要通过端口5672连接，而不是15672，更改端口，转发，一切正常。