Python并发编程

import argparse
import redis
from tqdm import tqdm
from multiprocessing import Pool, Process

parser = argparse.ArgumentParser(description='PyTorch MM Training')
parser.add_argument('--port', default=6379, type=str, help="port id")
parser.add_argument('--file', default=None, type=str, help="paths for train instances")
args = parser.parse_args()

host = '127.0.0.1'
port = args.port

# r = redis.Redis(host=host, port=port)

train_file = args.file

def worker_i(train_file, i, nums=16):
    r = redis.Redis(host=host, port=port)
    k = 0

    if i == 0:
        iter_ = tqdm(open(train_file, 'r'))
    else:
        iter_ = open(train_file, 'r')

    for info in iter_:
        if k % nums == i: # 将多个任务分成nums份
            r.set(k, info.strip())
        k += 1
process_list = []

nums = 8
for i in range(nums):
    p = Process(target=worker_i, args=(train_file, i, nums))
    p.start()
    process_list.append(p)

for i in process_list:
    p.join()

print('写入完成')

import threading

def craw(url):  # 1. 准备一个函数
	r = requests.get(url)
	print(len(r.text))

def multi_thread():
	threads = []  # 2. 准备一个线程list
	for url in urls:
		threads.append(  # 3. 创建线程
			threading.Thread(target=craw, args=(url,))
		)

	for thread in threads:
		thread.start()  # 4. 启动线程

	for thread in threads:
		thread.join()  # 5. 等待结束

from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as pool:
		result = pool.map(craw, urls)
		# map的结果和入参的顺序对应的
		for result in results:
				print(result)

from concurrent.futures import ThreadPoolExecutor, as_completed
with ThreadPoolExecutor() as pool:
		# 使用 dict 可以知道future对应的入参
		futures = {pool.submit(craw, url): url
								for url in urls}
		# 方式1: 结果仍然按顺序
		for future, url in futures.items():
				print(url, future.result())

		# 方式2: 使用 as_completed 顺序是不定的
		for future in as_completed(futures):  # 注：字典的遍历是遍历key
				url = futures[future]
				print(url, future.result())

import ctypes
import time
from multiprocessing import Process, RawValue

class CountdownTask:
    def __init__(self):
        self._running = RawValue(ctypes.c_bool, True)
    def terminate(self):
        self._running.value = False
    def run(self, n):
        while self._running.value and n > 0:
            print('T-minus', n)
            n -= 1
            time.sleep(5)

c = CountdownTask()
t = Process(target=c.run, args=(10,))
t.start()
c.terminate() # Signal termination
t.join() # Wait for actual termination (if needed)

import asyncio

# 获取事件循环（超级循环）
loop = asyncio.get_event_loop()

# 定义协程
async def myfunc(url):
		await get_url(url)

# 创建task列表
tasks = [loop.create_task(myfunc(url))
					for url in urls]

# 执行爬虫事件列表
loop.run_until_complete(asyncio.wait(tasks))

import asyncio, aiohtttp

async def async_craw(url):
		async with aiohttp.ClientSession() as sess:
				async with sess.get(url) as resp:
						result = await resp.text()
						print(url, len(result))

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

import asyncio, aiohtttp

semaphore = asyncio.Semaphore(10)
async def async_craw(url):
		async with semaphore:  # here
				async with aiohttp.ClientSession() as sess:
						async with sess.get(url) as resp:
								result = await resp.text()
								print(url, len(result))

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

# 方法一：使用with
sem = asyncio.Semaphore(10)

# ...later
async with sem:
		# work with shared resource

# 方法二：手动acquire、release
sem = asyncio.Semaphore(10)

# ...later
await sem.acquire()
try:
		# work with shared resource
finally:
		sem.release()