高并发场景下实时语音合成的性能优化

#!/usr/bin/env python3
# Copyright (C) Alibaba Group. All Rights Reserved.
# MIT License (https://opensource.org/licenses/MIT)

import os
import time
import threading

import dashscope
from dashscope.audio.tts_v2 import *


USE_CONNECTION_POOL = True
text_to_synthesize = [
  '第一句、欢迎使用阿里巴巴语音合成服务。',
  '第二句、欢迎使用阿里巴巴语音合成服务。',
  '第三句、欢迎使用阿里巴巴语音合成服务。',
]
connectionPool = None
if USE_CONNECTION_POOL:
  print('creating connection pool')
  start_time = time.time() * 1000
  connectionPool = SpeechSynthesizerObjectPool(max_size=3)
  end_time = time.time() * 1000
  print('connection pool created, cost: {} ms'.format(end_time - start_time))

def init_dashscope_api_key():
  if 'DASHSCOPE_API_KEY' in os.environ:
    dashscope.api_key = os.environ[
      'DASHSCOPE_API_KEY']  # load API-key from environment variable DASHSCOPE_API_KEY
  else:
    dashscope.api_key = 'sk-xxx'  # set API-key manually


def synthesis_text_to_speech_and_play_by_streaming_mode(text, task_id):
  global USE_CONNECTION_POOL, connectionPool

  complete_event = threading.Event()

  # Define a callback to handle the result

  class Callback(ResultCallback):
    def on_open(self):
      # when using object pool, on_open will be called after task start
      self.file = open(f'result_{task_id}.mp3', 'wb')
      print(f'[task_{task_id}] start')

    def on_complete(self):
      print(f'[task_{task_id}] speech synthesis task complete successfully.')
      complete_event.set()

    def on_error(self, message: str):
      print(f'[task_{task_id}] speech synthesis task failed, {message}')

    def on_close(self):
      # when using object pool, on_close will be called after task finished
      print(f'[task_{task_id}] finished')

    def on_event(self, message):
      pass

    def on_data(self, data: bytes) -> None:
      # save audio to file
      self.file.write(data)

  # Call the speech synthesizer callback
  synthesizer_callback = Callback()

  # Initialize the speech synthesizer
  if USE_CONNECTION_POOL:
    speech_synthesizer = connectionPool.borrow_synthesizer(
      model='cosyvoice-v3-flash',
      voice='longanyang',
      seed=12382,
      callback=synthesizer_callback
    )
  else:
    speech_synthesizer = SpeechSynthesizer(model='cosyvoice-v3-flash',
                                           voice='longanyang',
                                           seed=12382,
                                           callback=synthesizer_callback)
  try:
    speech_synthesizer.call(text)
  except Exception as e:
    print(f'[task_{task_id}] speech synthesis task failed, {e}')
    if USE_CONNECTION_POOL:
      # close the synthesizer connection manually if task failed when using connection pool.
      speech_synthesizer.close()
    return

  print('[task_{}] Synthesized text: {}'.format(task_id, text))
  complete_event.wait()
  print('[task_{}][Metric] requestId: {}, first package delay ms: {}'.format(
    task_id,
    speech_synthesizer.get_last_request_id(),
    speech_synthesizer.get_first_package_delay()))
  if USE_CONNECTION_POOL:
    connectionPool.return_synthesizer(speech_synthesizer)


# main function
if __name__ == '__main__':
  init_dashscope_api_key()
  task_thread_list = []
  for task_id in range(3):
    thread = threading.Thread(
      target=synthesis_text_to_speech_and_play_by_streaming_mode,
      args=(text_to_synthesize[task_id], task_id))
    task_thread_list.append(thread)

  for task_thread in task_thread_list:
    task_thread.start()

  for task_thread in task_thread_list:
    task_thread.join()

  if USE_CONNECTION_POOL:
    connectionPool.shutdown()

import com.alibaba.dashscope.audio.tts.SpeechSynthesisResult;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisAudioFormat;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisParam;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesizer;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.alibaba.dashscope.utils.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * 您需要在项目中引入org.apache.commons.pool2和DashScope相关的包。
 *
 * DashScope SDK 2.16.6及后续版本针对高并发场景进行了优化，
 * DashScope SDK 2.16.6之前的版本不推荐在高并发场景下使用。
 *
 * 在对TTS服务进行高并发调用之前，
 * 请通过以下环境变量配置连接池的相关参数。
 *
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST
 * DASHSCOPE_CONNECTION_POOL_SIZE
 */

class SpeechSynthesizerObjectFactory
    extends BasePooledObjectFactory<SpeechSynthesizer> {
  public SpeechSynthesizerObjectFactory() {
    super();
  }
  @Override
  public SpeechSynthesizer create() throws Exception {
    return new SpeechSynthesizer();
  }

  @Override
  public PooledObject<SpeechSynthesizer> wrap(SpeechSynthesizer obj) {
    return new DefaultPooledObject<>(obj);
  }
}

class CosyvoiceObjectPool {
  public static GenericObjectPool<SpeechSynthesizer> synthesizerPool;
  public static String COSYVOICE_OBJECTPOOL_SIZE_ENV = "COSYVOICE_OBJECTPOOL_SIZE";
  public static int DEFAULT_OBJECT_POOL_SIZE = 500;
  private static Lock lock = new java.util.concurrent.locks.ReentrantLock();
  public static int getObjectivePoolSize() {
    try {
      Integer n = Integer.parseInt(System.getenv(COSYVOICE_OBJECTPOOL_SIZE_ENV));
      System.out.println("Using Object Pool Size In Env: "+ n);
      return n;
    } catch (NumberFormatException e) {
      System.out.println("Using Default Object Pool Size: "+ DEFAULT_OBJECT_POOL_SIZE);
      return DEFAULT_OBJECT_POOL_SIZE;
    }
  }
  public static GenericObjectPool<SpeechSynthesizer> getInstance() {
    lock.lock();
    if (synthesizerPool == null) {
      // 您可以在这里设置对象池的大小。或在环境变量COSYVOICE_OBJECTPOOL_SIZE中设置。
      // 建议设置为服务器最大并发连接数的1.5到2倍。
      int objectPoolSize = getObjectivePoolSize();
      SpeechSynthesizerObjectFactory speechSynthesizerObjectFactory =
          new SpeechSynthesizerObjectFactory();
      GenericObjectPoolConfig<SpeechSynthesizer> config =
          new GenericObjectPoolConfig<>();
      config.setMaxTotal(objectPoolSize);
      config.setMaxIdle(objectPoolSize);
      config.setMinIdle(objectPoolSize);
      synthesizerPool =
          new GenericObjectPool<>(speechSynthesizerObjectFactory, config);
    }
    lock.unlock();
    return synthesizerPool;
  }
}

class SynthesizeTaskWithCallback implements Runnable {
  String[] textArray;
  String requestId;
  long timeCost;
  public SynthesizeTaskWithCallback(String[] textArray) {
    this.textArray = textArray;
  }
  @Override
  public void run() {
    SpeechSynthesizer synthesizer = null;
    long startTime = System.currentTimeMillis();
    // if recv onError
    final boolean[] hasError = {false};
    try {
      class ReactCallback extends ResultCallback<SpeechSynthesisResult> {
        ReactCallback() {}

        @Override
        public void onEvent(SpeechSynthesisResult message) {
          if (message.getAudioFrame() != null) {
            try {
              byte[] bytesArray = message.getAudioFrame().array();
              System.out.println("收到音频，音频文件流length为：" + bytesArray.length);
            } catch (Exception e) {
              throw new RuntimeException(e);
            }
          }
        }

        @Override
        public void onComplete() {}

        @Override
        public void onError(Exception e) {
          System.out.println(e.getMessage());
          e.printStackTrace();
          hasError[0] = true;
        }
      }

      SpeechSynthesisParam param =
          SpeechSynthesisParam.builder()
              .model("cosyvoice-v3-flash")
              .voice("longanyang")
              // 若没有配置环境变量，请用千问云 API Key将下行替换为：.apiKey("sk-xxx")
              .apiKey(System.getenv("DASHSCOPE_API_KEY"))
              .format(SpeechSynthesisAudioFormat
                  .MP3_22050HZ_MONO_256KBPS) // 流式合成使用PCM或者MP3
              .build();

      try {
        synthesizer = CosyvoiceObjectPool.getInstance().borrowObject();
        synthesizer.updateParamAndCallback(param, new ReactCallback());
        for (String text : textArray) {
          synthesizer.streamingCall(text);
        }
        Thread.sleep(20);
        synthesizer.streamingComplete(60000);
        requestId = synthesizer.getLastRequestId();
      } catch (Exception e) {
        System.out.println("Exception e: " + e.toString());
        hasError[0] = true;
      }
    } catch (Exception e) {
      hasError[0] = true;
      throw new RuntimeException(e);
    }
    if (synthesizer != null) {
      try {
        if (hasError[0] == true) {
          // 如果出现异常，则关闭连接并在对象池中禁用该对象。
          synthesizer.getDuplexApi().close(1000, "bye");
          CosyvoiceObjectPool.getInstance().invalidateObject(synthesizer);
        } else {
          // 如果任务正常结束，则归还对象。
          CosyvoiceObjectPool.getInstance().returnObject(synthesizer);
        }
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
      long endTime = System.currentTimeMillis();
      timeCost = endTime - startTime;
      System.out.println("[线程 " + Thread.currentThread() + "] 语音合成任务结束。耗时 " + timeCost + " ms, RequestId " + requestId);
    }
  }
}

@Slf4j
public class SynthesizeTextToSpeechWithCallbackConcurrently {
  public static void checkoutEnv(String envName, int defaultSize) {
    if (System.getenv(envName) != null) {
      System.out.println("[ENV CHECK]: " + envName + " "
          + System.getenv(envName));
    } else {
      System.out.println("[ENV CHECK]: " + envName
          + " Using Default which is " + defaultSize);
    }
  }

  public static void main(String[] args)
      throws InterruptedException, NoApiKeyException {
    // Check for connection pool env
    checkoutEnv("DASHSCOPE_CONNECTION_POOL_SIZE", 32);
    checkoutEnv("DASHSCOPE_MAXIMUM_ASYNC_REQUESTS", 32);
    checkoutEnv("DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST", 32);
    checkoutEnv(CosyvoiceObjectPool.COSYVOICE_OBJECTPOOL_SIZE_ENV, CosyvoiceObjectPool.DEFAULT_OBJECT_POOL_SIZE);

    int runTimes = 3;
    // Create the pool of SpeechSynthesis objects
    ExecutorService executorService = Executors.newFixedThreadPool(runTimes);

    for (int i = 0; i < runTimes; i++) {
      // Record the task submission time
      LocalDateTime submissionTime = LocalDateTime.now();
      executorService.submit(new SynthesizeTaskWithCallback(new String[] {
          "床前明月光，", "疑似地上霜。", "举头望明月，", "低头思故乡。"}));
    }

    // Shut down the ExecutorService and wait for all tasks to complete
    executorService.shutdown();
    executorService.awaitTermination(1, TimeUnit.MINUTES);
    System.exit(0);
  }
}

// 关闭连接
synthesizer.getDuplexApi().close(1000, "bye");
// 在对象池中废弃出现异常的synthesizer
CosyvoiceObjectPool.getInstance().invalidateObject(synthesizer);