Pythonでの同時リクエスト

サーバー側のテストでは、サービス側のビジネス機能とAPIの互換性を考慮することに加えて、サーバー側の安定性と、高い同時要求の下でのサーバー側の伝送容量を考慮する必要があります。並行性の数と特定の応答時間の要件に関しては、実際には、各製品の形式が異なり、標準的な用語を使用して統一することは困難です。テスト対象のコンポーネントが直面しているビジネスフォームによって異なります。ビジネスフォームがめったに使用されない製品である場合、実際にはパフォーマンスの要件はありません。したがって、これは、テスト対象のコンポーネントのアーキテクチャ設計、負荷の量、およびビジネス目標に基づいている必要があります。この記事では、主にPythonを使用して、同時リクエストの簡単なテストコードを記述します。

Pythonのコンカレントプログラミングモデルでは、関係する主なポイントは、スレッドとプロセス、および対応するコルーチンです。そして、イナゴは主にマイクロスレッドとして理解できるコロチンに基づいて設計されています。 IO集約型およびCPU集約型では、IO集約型の場合は、マルチスレッドを使用して効率を上げることをお勧めします。CPU集約型の場合は、マルチプロセス方式を使用して効率を上げることをお勧めします。この記事では、主にIOの集中度に基づく方法、つまりマルチスレッドについて説明します。スレッドを開始する方法は非常に簡単です。機能プログラミングまたはオブジェクト指向プログラミングを使用できます。次の特定のケースコード、機能方法を参照してください。

from threading import  Thread
import  time as t
import  random

def job(name):print('私は{0},仕事を始めたい'.format(name))if __name__ =='__main__':
 t=Thread(target=job,args=('Li Si',))
 t.start()print('メインスレッド実行の終了')

オブジェクト指向のアプローチ：

from threading import  Thread
import  time as t
import  random

classJob(Thread):
 def __init__(self,name):super().__init__()
  self.name=name

 def run(self)-> None:print('私は{0},仕事を始めたい'.format(self.name))if __name__ =='__main__':
 t=Job('Li Si')
 t.start()print('メインスレッドプログラム実行の終了')

実際、Threadクラスでは、テストされた関数の戻り値は返されません。つまり、APIインターフェイスをテストするときに、テストされたインターフェイスのステータスコード、要求の応答時間、および応答データを取得する必要があります。 Threadクラスを再継承した後、run（）メソッドを書き直して、テストした関数で期待されるデータを取得します。具体的なケースコードは次のとおりです。

#! coding:utf-8from threading import  Thread

classThreadTest(Thread):
 def __init__(self,func,args=()):'''
  : param func:テスト中の機能
  : param args:テストされている関数の戻り値
      '''
  super(ThreadTest,self).__init__()
  self.func=func
  self.args=args

 def run(self)-> None:
  self.result=self.func(*self.args)

 def getResult(self):try:return self.result
  except BaseException as e:return e.args[0]

ここでは、Baiduホームページのテストをケースとして取り上げ、同時リクエストを受信した後、同時リクエストの応答時間とステータスコードを取得し、応答時間に応じて中央値などのデータを取得します。具体的な完全なケースコードは次のとおりです。

#! /usr/bin/env python
#! coding:utf-8from threading import  Thread
import  requests
import  matplotlib.pyplot as plt
import  datetime
import  time
import  numpy as np
import  json

classThreadTest(Thread):
 def __init__(self,func,args=()):'''
  : param func:テスト中の機能
  : param args:テストされている関数の戻り値
      '''
  super(ThreadTest,self).__init__()
  self.func=func
  self.args=args

 def run(self)-> None:
  self.result=self.func(*self.args)

 def getResult(self):try:return self.result
  except BaseException as e:return e.args[0]

def baiDu(code,seconds):'''
 : param code:ステータスコード
 : param seconds:応答時間を要求する
 : return:'''
 r=requests.get(url='http://www.baidu.com/')
 code=r.status_code
 seconds=r.elapsed.total_seconds()return code,seconds

def calculationTime(startTime,endTime):'''2回の差を計算します。単位は秒です。'''return(endTime-startTime).seconds

def getResult(seconds):'''サーバーの応答時間情報を取得します'''
 data={'Max':sorted(seconds)[-1],'Min':sorted(seconds)[0],'Median':np.median(seconds),'99%Line':np.percentile(seconds,99),'95%Line':np.percentile(seconds,95),'90%Line':np.percentile(seconds,90)}return data

def highConcurrent(count):'''
 高い同時リクエストをサーバーに送信する
 : param cout:並行性
 : return:'''
 startTime=datetime.datetime.now()
 sum=0
 list_count=list()
 tasks=list()
 results =list()
 # 失敗したメッセージ
 fails=[]
 # 成功したタスクの数
 success=[]
 codes =list()
 seconds =list()for i inrange(1,count):
  t=ThreadTest(baiDu,args=(i,i))
  tasks.append(t)
  t.start()for t in tasks:
  t.join()if t.getResult()[0]!=200:
   fails.append(t.getResult())
  results.append(t.getResult())

 endTime=datetime.datetime.now()for item in results:
  codes.append(item[0])
  seconds.append(item[1])for i inrange(len(codes)):
  list_count.append(i)

 # 視覚的なトレンドチャートを生成する
 fig,ax=plt.subplots()
 ax.plot(list_count,seconds)
 ax.set(xlabel='number of times', ylabel='Request time-consuming',
   title='olap continuous request response time (seconds)')
 ax.grid()
 fig.savefig('olap.png')
 plt.show()for i in seconds:
  sum+=i
 rate=sum/len(list_count)
 # print('\n合計期間:\n',endTime-startTime)
 totalTime=calculationTime(startTime=startTime,endTime=endTime)if totalTime<1:
  totalTime=1
 # スループットの計算
 try:
  throughput=int(len(list_count)/totalTime)
 except Exception as e:print(e.args[0])getResult(seconds=seconds)
 errorRate=0iflen(fails)==0:
  errorRate=0.00else:
  errorRate=len(fails)/len(tasks)*100
 throughput=str(throughput)+'/S'
 timeData=getResult(seconds=seconds)
 dict1={'スループット':throughput,'平均応答時間':rate,'反応時間':timeData,'エラー率':errorRate,'リクエストの総数':len(list_count),'失敗の数':len(fails)}return  json.dumps(dict1,indent=True,ensure_ascii=False)if __name__ =='__main__':print(highConcurrent(count=1000))

上記のコードが実行されると、視覚的な要求の応答時間などの情報が形成され、実行後に次の情報が表示されます。

{" スループット":"500/S","平均応答時間":0.08835436199999998,"反応時間":{"Max":1.5547,"Min":0.068293,"Median":0.0806955,"99%Line":0.12070111,"95%Line":0.10141509999999998,"90%Line":0.0940216},"エラー率":0.0,"リクエストの総数":1000,"失敗の数":0}

ご清聴ありがとうございました。今後も更新し、具体的に呼び出すことができるインターフェースとして提供していきます。コードを使用すると、サーバーの伝送容量と、作業中にタイムアウトがあるかどうかをすばやく確認できます。