from __future__ import print_function
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor
from volcengine import visual
from volcengine.visual.VisualService import VisualService
def send_request(visual_service, req_body, request_count, result_queue):
start_time = time.time()
resp = visual_service.cv_process(req_body)
'request_index': request_count,
'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'),
'elapsed': end_time - start_time
'request_index': request_count,
'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
def __init__(self, rate):
self.last_time = time.time()
self.lock = threading.Lock()
elapsed = now - self.last_time
new_tokens = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
def get_token(self, block=True):
block: 如果为True,则阻塞直到获取到令牌;否则立即返回是否获取到令牌
wait_time = (1 - self.tokens) / self.rate
def process_results(result_queue, count_lock, success_count, failure_count):
result = result_queue.get(block=True, timeout=0.1)
print(f"请求 {result['request_index']} 失败: {result['error']} 时间: {result['time']}")
print(f"请求 {result['request_index']} 成功, 耗时: {result['elapsed']:.4f}秒, 时间: {result['time']}")
def maintain_qps(ak, sk, req_body, target_qps=3, duration_seconds=None, max_workers=100):
ak: AccessKey of your account
sk: SecretKey of your account
duration_seconds: 运行持续时间(秒),如果为None则一直运行
max_workers: 最大线程池工作线程数,默认100
visual_service = VisualService()
visual_service.set_ak(ak)
visual_service.set_sk(sk)
start_time = time.time()
count_lock = threading.Lock()
executor = ThreadPoolExecutor(max_workers=max_workers)
token_bucket = TokenBucket(target_qps)
result_thread = threading.Thread(
args=(result_queue, count_lock, success_count, failure_count)
result_thread.daemon = True
print(f"开始以精确的 {target_qps} QPS 频率发送请求...")
last_stats_time = start_time
while duration_seconds is None or time.time() - start_time < duration_seconds:
token_bucket.get_token(block=True)
current_request_index = request_count
visual_service, req_body,
current_request_index, result_queue
current_time = time.time()
if current_time - last_stats_time >= 10:
current_qps = request_count / (current_time - start_time)
print(f"当前平均QPS: {current_qps:.2f}, 已发送请求数: {request_count}, 成功: {success_count[0]}, 失败: {failure_count[0]}")
last_stats_time = current_time
except KeyboardInterrupt:
result_thread.join(timeout=2)
executor.shutdown(wait=False)
total_time = time.time() - start_time
print(f"总共发送了 {request_count} 个请求,成功: {success_count[0]}, 失败: {failure_count[0]}, 耗时 {total_time:.2f} 秒")
print(f"平均QPS: {request_count / total_time:.2f}")
if __name__ == '__main__':
"req_key": "high_aes_general_v21_L",
max_workers = target_qps * 30 * 1.2
maintain_qps(ak, sk, req_body, target_qps, max_workers=max_workers)