博客大纲

from concurrent.futures import ThreadPoolExecutor
import requests
import time

start = time.time()
url = 'https://blog.towavephone.com'

def task(name):
  try:
     result = requests.get(url)
     print(name + 'ping ' + url + result.text[0:20])
  except Exception as e:
     print('except:', e)

if __name__ == "__main__":
  pool = ThreadPoolExecutor(100)

  for i in range(1, 500):
     pool.submit(task, "请求 %s：" % i)

  pool.shutdown()
  end = time.time()
  print("主线程结束，花费时间为：", end - start)

import queue
import threading
import time
import requests

start = time.time()
url = 'https://blog.towavephone.com'

def task(que):
  while not que.empty():
     name = que.get()
     try:
           result = requests.get(url)
           print(name + 'ping ' + url + result.text[0:20])
     except Exception as e:
           print('except:', e)

q = queue.Queue()
for i in range(1, 500):
  q.put('请求 %s：' % i)

max_thread = 100
ts = []
for i in range(max_thread):
  t = threading.Thread(target=task, args=(q,))
  t.start()
  ts.append(t)
for t in ts:
  t.join()

end = time.time()
print("主线程结束，花费时间为：", end - start)

import asyncio
import time
import requests
import httpx
import uvloop

start = time.time()
url = 'https://blog.towavephone.com'

sem = asyncio.Semaphore(100)

async def task(name, client):
  async with sem:
     try:
           result = await client.get(url)
           print(name + 'ping ' + url + result.text[0:20])
     except Exception as e:
           print('except:', e)

async def main():
  async with httpx.AsyncClient() as client:
     # 获取 EventLoop:
     tasks = []
     for i in range(1, 500):
           tasks.append(task('请求 %s：' % i, client))
     await asyncio.gather(*tasks)
     # 执行 coroutine

uvloop.install()
asyncio.run(main())

end = time.time()
print("主线程结束，花费时间为：", end - start)

import asyncio
import time
import httpx
import random
# import uvloop

start = time.time()
url = 'https://blog.towavephone.com'

sem = asyncio.Semaphore(1)

count = 0
is_success = True

async def task(name, client):
  async with sem:
     global count
     global is_success

     if not is_success:
           return
     for i in range(0, 3):
           try:
              result = await client.get(url)
              print(name + 'ping ' + url + result.text[0:20], count)
              if i == 2 and random.randint(0, 1):
                 print('------------' + name)
                 is_success = False
                 break
                 # raise asyncio.CancelledError('请求' + name + '取消')
           except Exception as e:
              print('except:', e)

     count = count + 1

     # return result.text[0:4]

async def main():
  async with httpx.AsyncClient() as client:
     # 获取 EventLoop:
     tasks = []
     for i in range(1, 100):
           tasks.append(task('请求 %s：' % i, client))
     result = await asyncio.gather(*tasks)
     print(result)
# uvloop.install()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
end = time.time()
print("主线程结束，花费时间为：", end - start)

import matplotlib.pyplot as plt
import numpy as np
import numpy.polynomial.polynomial as poly

# https://stackoverflow.com/questions/49037902/how-to-interpolate-a-line-between-two-other-lines-in-python
def interpolate(a1, a2, poly_deg=3, n_points=10, plot=True):
  min_a1_x, max_a1_x = min(a1[:, 0]), max(a1[:, 0])
  new_a1_x = np.linspace(min_a1_x, max_a1_x, n_points)
  a1_coefs = poly.polyfit(a1[:, 0], a1[:, 1], poly_deg)
  new_a1_y = poly.polyval(new_a1_x, a1_coefs)

  min_a2_x, max_a2_x = min(a2[:, 0]), max(a2[:, 0])
  new_a2_x = np.linspace(min_a2_x, max_a2_x, n_points)
  a2_coefs = poly.polyfit(a2[:, 0], a2[:, 1], poly_deg)
  new_a2_y = poly.polyval(new_a2_x, a2_coefs)

  midx = [np.mean([new_a1_x[i], new_a2_x[i]]) for i in range(n_points)]
  midy = [np.mean([new_a1_y[i], new_a2_y[i]]) for i in range(n_points)]

  if plot:
     plt.plot(a1[:, 0], a1[:, 1], c='black')
     plt.plot(a2[:, 0], a2[:, 1], c='black')
     plt.plot(midx, midy, '--', c='black')
     plt.show()

  return [[x, y] for x, y in zip(midx, midy)]

a1 = np.array(
  [
     [
           113.38226806372708,
           23.162344275038354
     ],
     [
           113.3822851628109,
           23.163151871405752
     ],
     [
           113.38228818029653,
           23.163380278265958
     ],
     [
           113.3822873421064,
           23.163461918002366
     ],
     [
           113.38228516281187,
           23.163597788651938
     ],
     [
           113.38227460161451,
           23.16384656353636
     ],
     [
           113.38225507178088,
           23.16411109639837
     ],
     [
           113.3822487853537,
           23.164175637052246
     ]
  ]
)
a2 = np.array(
  [
     [
           113.38236629963569,
           23.16418460568882
     ],
     [
           113.38238390163173,
           23.163970028969388
     ],
     [
           113.38239580393373,
           23.16377364097953
     ],
     [
           113.38239580393373,
           23.16377364097953
     ],
     [
           113.38239823468551,
           23.16371362655323
     ],
     [
           113.38239823468551,
           23.16371362655323
     ],
     [
           113.3824045211124,
           23.163508018469752
     ],
     [
           113.3824045211118,
           23.16320585086209
     ],
     [
           113.3824045211118,
           23.16320585086209
     ],
     [
           113.38240284473103,
           23.163134185590337
     ],
     [
           113.38240284473103,
           23.163134185590337
     ],
     [
           113.38239907287435,
           23.162987502285524
     ],
     [
           113.38239907287435,
           23.162987502285524
     ],
     [
           113.38239748031269,
           23.162955483415537
     ],
     [
           113.38239748031269,
           23.162955483415537
     ],
     [
           113.38238549419019,
           23.162341592829492
     ]
  ]
)

result = interpolate(a1, a2)
print('result', result)