## Musicca

一个广受全球用户喜爱的音乐学习平台之一，使用免费且充满乐趣音乐学习方式帮助全球用户学习音乐。

网站提供课程、练习和互动工具，鼓励音乐学习者在校和家中都能按照自己的节奏来学习，帮助学习者理解音乐、学会读谱和演奏乐器，而且通过在学习过程中加入游戏的元素，使学习音乐的过程变得有趣。

除此之外网站还提供大量的在线虚拟音乐工具，例如钢琴、吉他、贝斯等等，可以在线模拟真实的乐器使用。

## 网站地址

Musicca：[https://www.musicca.com/zh](https://www.musicca.com/zh)


Musicca-免费且充满乐趣音乐学习工具平台

很多时候，数据读写不一定是文件，也可以在内存中读写。

## StringIO

StringIO顾名思义就是在内存中读写str。

要把str写入StringIO，我们需要先创建一个StringIO，然后，像文件一样写入即可：

    from io import StringIO
    s = "abcdefg123"
    sio = StringIO(s)  # sio就是可变字符串
    print(sio.getvalue())  # abcdefg123
    sio.seek(7)  # 指针移动到索引7这个位置
    sio.write("666")
    print(sio.getvalue())  # abcdefg666

要读取StringIO，可以用一个str初始化StringIO，然后，像读文件一样读取：

    f = StringIO("hello\nworld")
    while True:
    s = f.readline()
    if s == "":
    break
    print(s.strip())

## BytesIO

StringIO操作的只能是str，如果要操作二进制数据，就需要使用BytesIO。

BytesIO实现了在内存中读写bytes，我们创建一个BytesIO，然后写入一些bytes：

    from io import BytesIO
    f = BytesIO()
    f.write("你好".encode("utf-8"))
    print(f.getvalue())

请注意，写入的不是str，而是经过UTF-8编码的bytes。

和StringIO类似，可以用一个bytes初始化BytesIO，然后，像读文件一样读取：

    f = BytesIO(b"xe4xbdxa0xe5xa5xbd")
    print(f.read())

## 小结

StringIO和BytesIO是在内存中操作str和bytes的方法，使得和读写文件具有一致的接口。


Python io模块StringIO和BytesIO

对操作系统来说，一个任务就是一个进程（Process），比如打开一个浏览器就是启动一个浏览器进程，打开两个记事本就启动了两个记事本进程。

有些进程不止同时干一件事，比如Word，它可以同时进行打字、拼写检查、打印等事情。在一个进程内部，要同时干多件事，就需要同时运行多个“子任务”，我们把进程内的这些“子任务”称为线程（Thread）。

进程之间是内存隔离的，即不同的进程拥有各自的内存空间。这就类似于不同的公司拥有不同的办公场所。

线程之间是内存共享的，线程是属于进程的，一个进程内的多个线程之间是共享这个进程所拥有的内存空间的。这就好比，公司员工之间共享公司的办公场所。

进程是操作系统分配资源的最小单位，而线程是程序执行（CPU调度）的最小单位。

## 进程

multiprocessing模块是跨平台版本的多进程模块，提供了一个Process类来代表一个进程对象。 import os from multiprocessing import Process

# 子进程要执行的代码

def run_proc(name):
print("子进程 %s (%s)" % (os.getpid(), name))
print("父进程 %s" % os.getppid())  # getppid 获取当前父进程编号

if **name** == "**main**":
# getpid 获取当前进程编号
print("父进程 %s" % os.getpid())
p = Process(target=run_proc, args=("test",))
p2 = Process(target=run_proc, kwargs={"name": "test2"})
print("Child process start")
p.start()
p2.start()
p.join()
p2.join()
print("Child process end")

创建子进程时，只需要传入一个执行函数和参数，创建一个Process实例，用start()方法（仅仅只是给操作系统发送了一个信号）启动。

join()方法：主进程等待子进程结束后再继续往下运行，通常用于进程间的同步。

进程间是不共享全局变量的。实际上创建一个子进程就是把主进程的资源进行拷贝产生了一个新进程，这里主进程和子进程是互相独立的。

### 守护主进程

实际上，主进程会等待所有子进程执行结束再结束。通过设置守护主进程，来保证主进程执行完毕，子进程随之销毁。 import time from multiprocessing import Process

def work():
for i in range(10):
print("工作中...")
time.sleep(0.2)

if **name** == "**main**":
work_process = Process(target=work)
work_process.daemon = True  # 方式一 设置守护主进程
work_process.start()
time.sleep(1)
# work_process.terminate()  # 方式二 直接销毁子进程
print("主进程执行结束")

守护进程必须是子进程，且不允许再创建新的子进程。主进程结束，守护进程也会随之结束。守护进程必须在start之前，start之后不能再设置daemon。

### 互斥锁

from multiprocessing import Process, Lock import time

def task(name, mutex):
mutex.acquire()  # 上锁
print(f"{name} 1")
time.sleep(1)
print(f"{name} 2")
mutex.release()  # 解锁
# with lock 写法
# with mutex:
#     print(f"{name} 1")
#     time.sleep(1)
#     print(f"{name} 2")

if **name** == "**main**":
mutex = Lock()
for i in range(3):
p = Process(target=task, args=(f"进程{i}", mutex))
p.start()

Lock不能多次上锁，推荐使用RLock可以多次上锁。

### 进程池Pool

如果要启动大量的子进程，可以用进程池的方式批量创建子进程： import os import time import random from multiprocessing import Pool

def long_time_task(name):
print("Run task %s (%s)..." % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print("Task %s runs %0.2f seconds." % (name, (end - start)))

if **name** == "**main**":
print("Parent process %s." % os.getpid())
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print("Waiting for all subprocesses done...")
p.close()
p.join()
print("All subprocesses done.")

对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了。

注意输出的结果，task 0，1，2，3是立刻执行的，而task 4要等待前面某个task完成后才执行，这是因为Pool的默认大小是CPU核数，这里设置为4最多同时执行4个进程。

如果改成`p = Pool(5)`就可以同时跑5个进程。

### 子进程

很多时候，子进程并不是自身，而是一个外部进程。创建了子进程后，还需要控制子进程的输入和输出。

subprocess模块可以非常方便地启动一个子进程，然后控制其输入和输出。

下面的例子演示了如何在Python代码中运行命令`nslookup www.python.org`，这和命令行直接运行的效果是一样的：

import subprocess print("$ nslookup www.python.org") r = subprocess.call(["nslookup", "www.python.org"]) print("Exit code:", r)

如果子进程还需要输入，则可以通过communicate()方法输入：

import subprocess print("$ nslookup") p = subprocess.Popen(["nslookup"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) output, err = p.communicate(b"set q=mxnpython.orgnexitn") print(output.decode("gbk"))  # Windows默认编码是gbk print("Exit code:", p.returncode)

上面的代码相当于在命令行执行命令nslookup，然后手动输入：

    set q=mx
    python.org
    exit

输入`set q=mx`检查某个DNS域名的MX记录，然后输入想要检查的DNS域名，输入之后，将显示DNS服务器的MX记录及对应的IP地址。

### 进程间通信

multiprocessing模块包装了底层机制，提供了Queue、Pipes等多种方式来交换数据。

以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：

import os import time import random from multiprocessing import Process, Queue # 写数据进程执行的代码 def write(q): print("Process to write: %s" % os.getpid()) for value in ["A", "B", "C"]: print("Put %s to queue..." % value) q.put(value) time.sleep(random.random()) # 读数据进程执行的代码 def read(q): print("Process to read: %s" % os.getpid()) while True: value = q.get(True) print("Get %s from queue." % value) if __name__ == "__main__": # 父进程创建Queue，并传给各个子进程： q = Queue() pw = Process(target=write, args=(q,)) pr = Process(target=read, args=(q,)) # 启动子进程pw，写入 pw.start() # 启动子进程pr，读取 pr.start() # 等待pw结束 pw.join() # pr进程里是死循环，无法等待其结束，只能强行终止。使用terminate终止进程，不会引起finally执行！ pr.terminate()

## 线程

Python的线程是真正的Posix Thread，而不是模拟出来的线程。利用Python的标准库threading，可以轻松实现多线程任务。

启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()开始执行：

import time import threading def sing(msg): n = 0 while n < 5: n += 1 print(f"{threading.current_thread().name} is running...{msg}") time.sleep(1) def dance(msg): n = 0 while n < 5: n += 1 print(f"{threading.current_thread().name} is running...{msg}") time.sleep(1) if __name__ == "__main__": # 以元组方式传参 sing_thread = threading.Thread(target=sing, args=("我在唱歌，啦啦啦",)) # 以字典方式传参 dance_thread = threading.Thread(target=dance, kwargs={"msg": "我在跳舞，呱呱呱"}) sing_thread.start() dance_thread.start()

任何进程默认就会启动一个线程，把该线程称为主线程，主线程又可以启动新的线程，Python的threading模块有个current_thread()函数，它永远返回当前线程的实例。

主线程实例的名字叫MainThread，子线程的名字可在创建时指定，如果不起名字Python会自动给线程命名为Thread-1 Thread-2 ...

### 守护主线程

实际上，主线程会等待所有子线程执行结束再结束。通过设置守护主线程，来保证主线程执行完毕，子线程随之销毁。 from threading import Thread import time

def task(name):
time.sleep(1)
print(f"你好，{name}")

if **name** == "**main**":
t = Thread(target=task, args=("张三",))
t.setDaemon(True)  # 守护线程，必须在 start() 之前设置
# t.daemon = True  # 另一种设置方法
t.start()
print("主线程")

### 锁Lock

多线程都在同个进程中，多线程使用的资源都是同一个进程中的资源，因此多线程间是共享全局变量的。因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。

来看看多个线程同时操作一个变量怎么把内容给改乱了：

import threading # 假定这是你的银行存款 balance = 0 def change_it(n): # 先存后取，结果应该为0 global balance balance = balance + n balance = balance - n def run_thread(n): for i in range(2000000): change_it(n) t1 = threading.Thread(target=run_thread, args=(5,)) t2 = threading.Thread(target=run_thread, args=(8,)) t1.start() t2.start() t1.join() t2.join() print(balance)

变量balance理论上结果应该为0，但是，由于线程的调度是由操作系统决定的，当t1、t2交替执行时，只要循环次数足够多，balance的结果就不一定是0了。

原因是因为高级语言的一条语句在CPU执行时是若干条语句，即使一个简单的计算`balance = balance + n`也分两步：

1. 计算balance + n，存入临时变量中
2. 将临时变量的值赋给balance

可看成是： x = balance + n balance = x因为修改balance需要多条语句，而执行这几条语句时，线程可能中断，从而导致多个线程把同一个对象的内容改乱了。

要确保balance计算正确，就要给change_it()上一把锁（互斥锁mutex），当某个线程开始执行change_it()时，该线程因为获得了锁，其他线程就不能同时执行change_it()，只能等待，直到锁被释放后，获得该锁以后才能改。由于锁只有一个，无论多少线程，同一时刻最多只有一个线程持有该锁，所以，不会造成修改的冲突。

创建一个锁通过threading.Lock()来实现：

import threading balance = 0 lock = threading.Lock() def change_it(n): # 先存后取，结果应该为0 global balance balance = balance + n balance = balance - n def run_thread(n): for i in range(2000000): # 先要获取锁 lock.acquire() try: # 放心地改吧 change_it(n) finally: # 改完了一定要释放锁，否则会产生死锁 lock.release() t1 = threading.Thread(target=run_thread, args=(5,)) t2 = threading.Thread(target=run_thread, args=(8,)) t1.start() t2.start() t1.join() t2.join() print(balance)

用try...finally模式来确保锁一定会被释放。也可以使用with lock模式：

import threading lock = threading.Lock() class Account: def __init__(self, balance): self.balance = balance def draw(account, amount): with lock: if account.balance >= amount: print("取钱成功") account.balance -= amount print("余额", account.balance) else: print("取钱失败，余额不足") account = Account(1000) t1 = threading.Thread(target=draw, args=(account, 600)) t2 = threading.Thread(target=draw, args=(account, 800)) t1.start() t2.start()

### GIL全局锁

Python的线程虽是真正的线程，但解释器执行代码时，有一个GIL锁：Global Interpreter Lock，任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都给上了锁，所以，多线程在Python中只能交替执行，即使100个线程跑在100核CPU上，也只能用到1个核。

所以，在Python中，可以使用多线程，但不要指望能有效利用多核。

不过，Python虽然不能利用多线程实现多核任务，但可以通过多进程实现多核任务。多个Python进程有各自独立的GIL锁，互不影响。

### ThreadLocal

在多线程环境下，每个线程都有自己的数据。一个线程使用自己的局部变量比使用全局变量好，因为局部变量只有线程自己能看见，不会影响其他线程，而全局变量的修改必须加锁。

但局部变量也有问题，就是在函数调用的时候，传递起来很麻烦。

ThreadLocal解决了参数在一个线程中各个函数之间互相传递的问题。一个ThreadLocal变量虽然是全局变量，但每个线程都只能读写自己线程的独立副本，互不干扰。

import threading # 创建全局ThreadLocal对象 local_school = threading.local() def process_student(): # 获取当前线程关联的student std = local_school.student print("Hello, %s (in %s)" % (std, threading.current_thread().name)) def process_thread(name): # 绑定ThreadLocal的student local_school.student = name process_student() t1 = threading.Thread(target=process_thread, args=("Alice",), name="Thread-A") t2 = threading.Thread(target=process_thread, args=("Bob",), name="Thread-B") t1.start() t2.start() t1.join() t2.join()

全局变量local_school是一个ThreadLocal对象，每个Thread对它都可以读写student属性，但互不影响。可以把local_school看成全局变量，但每个属性如local_school.student都是线程的局部变量，可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。

可以理解为全局变量local_school是一个dict，不但可以用local_school.student，还可以绑定其他变量，如local_school.teacher等。

ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。

### 在爬虫中的应用

import requests import os import bs4 import threading

os.makedirs("xkcd", exist_ok=True)  # store comics in ./xkcd

def downloadXkcd(startComic, endComic):
for urlNumber in range(startComic, endComic):
# Download the page.
print(f"Downloading page https://xkcd.com/{urlNumber}...")
res = requests.get(f"https://xkcd.com/{urlNumber}")
res.raise_for_status()

        soup = bs4.BeautifulSoup(res.text, "html.parser")
    
        # Find the URL of the comic image.
        comicElem = soup.select("#comic img")
        if comicElem == []:
            print("Could not find comic image.")
        else:
            comicUrl = comicElem[0].get("src")
            # Download the image.
            print(f"Downloading image {comicUrl}...")
        res = requests.get("https:" + comicUrl)
        res.raise_for_status()
    
        # Save the image to ./xkcd.
        imageFile = open(os.path.join("xkcd", os.path.basename(comicUrl)), "wb")
        for chunk in res.iter_content(100000):
            imageFile.write(chunk)
        imageFile.close()
    

downloadThreads = []  # a list of all the Thread objects
for i in range(0, 140, 10):  # loops 14 times, creates 14 threads
start = i
end = i + 9
if start == 0:
start = 1  # There is no comic 0, so set it to 1.
downloadThread = threading.Thread(target=downloadXkcd, args=(start, end))
downloadThreads.append(downloadThread)
downloadThread.start()

for downloadThread in downloadThreads:
downloadThread.join()
print("Done.")

### 多线程数据通信

queue.Queue可用于多线程间的、线程安全的数据通信。 import queue

q = queue.Queue()

# 添加元素

q.put("100")
q.put("200")

# 获取元素

item = q.get()
print(item)  # 100
item2 = q.get()
print(item2)  # 200

# 查看元素多少

print(q.qsize())  # 0

# 判断是否为空

print(q.empty())  # True

# 判断是否已满

print(q.full())  # False

### 信号量

from threading import Thread, Semaphore, currentThread import time import random

semaphore = Semaphore(3)

def task():
with semaphore:
print(currentThread().getName())
time.sleep(random.randint(1, 3))

if **name** == "**main**":
for i in range(10):
t = Thread(target=task)
t.start()


Python进程和线程

Python中**单线程**的异步编程模型称为协程。一般情况下，当程序处于IO操作的时候，线程都会处于阻塞状态。线程是CPU控制的，而协程是程序自身控制的，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级。

协程：当程序遇见了IO操作的时候，可以选择性的切换到其他任务上。在单线程条件下，微观上协程是一个任务一个任务的进行切换，宏观上看是多个任务一起在执行（多任务异步操作）。

要用在异步IO编程中依赖的库必须支持异步IO特性。

## asyncio

是Python 3.4之后引入的标准库，内置对异步IO的支持。其编程模型是一个消息循环，从asyncio模块中直接获取一个EventLoop的引用，然后把需要执行的协程扔到EventLoop中执行，就实现了异步IO。 import asyncio  # 导入异步编程库 import time

async def func1():
print("李小龙")
await asyncio.sleep(5)
print("李小龙")

async def func2():
print("李大龙")
await asyncio.sleep(3)
print("李大龙")

async def main():
tasks = [asyncio.create_task(func1()), asyncio.create_task(func2())]
await asyncio.wait(tasks)

if **name** == "**main**":
t1 = time.time()
asyncio.run(main())
t2 = time.time()
print(t2 - t1)

## 异步爬虫

爬虫中requests库不支持异步，需要用aiohttp。 import asyncio import aiohttp import time

async def async_craw(url):
print("craw url:", url)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
result = await resp.text()
print(f"craw url: {url}, {len(result)}")

urls = [f"https://news.cnblogs.com/n/page/{page}" for page in range(1, 10 + 1)]

loop = asyncio.get_event_loop()
tasks = [loop.create_task(async_craw(url)) for url in urls]

start = time.time()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
print("use time seconds:", end - start)

Python 3.7以上版本的写法：import asyncio import aiohttp import time urls = [f"https://news.cnblogs.com/n/page/{page}" for page in range(1, 10 + 1)] async def async_craw(url): print("craw url:", url) async with aiohttp.ClientSession() as session: async with session.get(url) as resp: result = await resp.text() print(f"craw url: {url}, {len(result)}") async def main(): tasks = [asyncio.create_task(async_craw(url)) for url in urls] asyncio.wait(tasks) if __name__ == "__main__": start = time.time() asyncio.run(main()) end = time.time() print("use time seconds:", end - start)

## 信号量Semaphore

信号量又称为旗语，是一个计数器，用来设置并发度。 import asyncio import aiohttp import time

semaphore = asyncio.Semaphore(5)  # 信号量 5

async def async_craw(url):
async with semaphore:
print("craw url:", url)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
result = await resp.text()
print(f"craw url: {url}, {len(result)}")

urls = [f"https://news.cnblogs.com/n/page/{page}" for page in range(1, 10 + 1)]

loop = asyncio.get_event_loop()
tasks = [loop.create_task(async_craw(url)) for url in urls]

start = time.time()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
print("use time seconds:", end - start)


Python协程单线程异步IO实现并发

Python 3.3版本之后自带venv，可直接代替virtualenv。

通过使用虚拟环境，可以：

- 避免库和依赖项之间的冲突
- 测试不同版本的库而不会影响系统范围的 Python 安装
- 维护干净、有序的开发环境
- 与其他人合作，同时确保库版本一致

## 使用Python虚拟环境

创建虚拟环境：`python -m venv myenv`。此命令会在当前目录中创建一个名为“myenv”的虚拟环境。-m选项指定运行的是venv模块，最后的myenv是创建的虚拟环境路径。

在 Windows 上：

- 激活虚拟环境：`.\myenv\Scripts\activate`
- 退出虚拟环境：`deactivate`

## 最佳实践和建议

当进入虚拟环境时，请牢记以下最佳做法：

**为每个项目创建一个虚拟环境**：每当启动一个新项目时，就创建一个新的虚拟环境。这可确保工作空间干净且隔离。

**使用需求文件**：要记录和管理项目的依赖项，请创建一个 requirements.txt 文件。可以使用`pip freeze > requirements.txt`生成它，然后使用`pip install -r requirements.txt`将它们安装在新环境中。

**激活和停用**：在开始项目之前，请务必激活相应的虚拟环境，并在完成后将其停用。这可以避免混淆和潜在冲突。

**版本控制**：与他人合作，请在版本控制系统中包含虚拟环境设置说明。这可确保每个人都使用相同的环境。

**升级 pip 和 setuptools**：创建新的虚拟环境时，最好将pip和setuptools升级到最新版本。这可确保使用的是最新的工具。

## 使用虚拟环境

pip install scrapy -i https://pypi.douban.com/simple/Scrapy爬虫框架依赖的库较多，如果出现报错，可采用wheel方式安装。推荐到 [https://www.lfd.uci.edu/~gohlke/pythonlibs/](https://www.lfd.uci.edu/~gohlke/pythonlibs/) 下载对应的wheel文件进行安装。

创建Scrapy项目：`scrapy startproject Test`

创建爬虫：`scrapy genspider spidername domain`，spidername是要创建的Spider名称，domain是允许爬取的域名。

运行爬虫：`scrapy crawl spidername`

调试Scrapy源码，可在项目根目录下新建main.py文件：

from scrapy.cmdline import execute import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) execute(["scrapy", "crawl", "spidername"])

然后调试main.py文件即可。

⚠️
注意：下面是virtualenv相关内容，不推荐使用了。
## 创建虚拟环境

pip install virtualenv virtualenv 环境名称

## 激活、退出虚拟环境

打开终端，进入创建好的虚拟环境 cd Scripts activate # 激活 deactivate # 退出


为什么在 Python 中使用虚拟环境？

这里是 aboss.top 的自留地。写了十几年野路子代码，攒了些摸爬滚打的经验，做成系列课程慢慢讲；也留了一小块地方，记记代码之外的胡思乱想。