高并发场景下实时语音识别的性能优化

package org.alibaba.bailian.example.examples;

import com.alibaba.dashscope.audio.asr.recognition.Recognition;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionParam;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionResult;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.alibaba.dashscope.utils.ApiKey;
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.io.FileInputStream;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * Before making high-concurrency calls to the ASR service,
 * please configure the connection pool size through following environment
 * variables.
 *
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS=2000
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST=2000
 * DASHSCOPE_CONNECTION_POOL_SIZE=2000
 *
 * The default is 32, and it is recommended to set it to 2 times the maximum
 * concurrent connections of a single server.
 */
public class Main {
  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 NoApiKeyException, InterruptedException {
    // 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(RecognitionObjectPool.RECOGNITION_OBJECTPOOL_SIZE_ENV, RecognitionObjectPool.DEFAULT_OBJECT_POOL_SIZE);

    int threadNums = 3;
    String currentDir = System.getProperty("user.dir");
    // Please replace the path with your audio source
    Path[] filePaths = {
        Paths.get(currentDir, "asr_example.wav"),
        Paths.get(currentDir, "asr_example.wav"),
        Paths.get(currentDir, "asr_example.wav"),
    };
    // Use ThreadPool to run recognition tasks
    ExecutorService executorService = Executors.newFixedThreadPool(threadNums);
    for (int i = 0; i < threadNums; i++) {
      executorService.submit(new RealtimeRecognizeTask(filePaths));
    }
    executorService.shutdown();
    // wait for all tasks to complete
    executorService.awaitTermination(10, TimeUnit.MINUTES);
    System.exit(0);
  }
}

class RecognitionObjectFactory extends BasePooledObjectFactory<Recognition> {
  public RecognitionObjectFactory() {
    super();
  }

  @Override
  public Recognition create() throws Exception {
    return new Recognition();
  }

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

class RecognitionObjectPool {
  public static GenericObjectPool<Recognition> recognitionGenericObjectPool;
  public static String RECOGNITION_OBJECTPOOL_SIZE_ENV =
      "RECOGNITION_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(RECOGNITION_OBJECTPOOL_SIZE_ENV));
      return n;
    } catch (NumberFormatException e) {
      return DEFAULT_OBJECT_POOL_SIZE;
    }
  }

  public static GenericObjectPool<Recognition> getInstance() {
    lock.lock();
    if (recognitionGenericObjectPool == null) {
      // You can set the object pool size here. or in environment variable
      // RECOGNITION_OBJECTPOOL_SIZE It is recommended to set it to 1.5 to 2
      // times your server's maximum concurrent connections.
      int objectPoolSize = getObjectivePoolSize();
      System.out.println("RECOGNITION_OBJECTPOOL_SIZE: "
          + objectPoolSize);
      RecognitionObjectFactory recognitionObjectFactory =
          new RecognitionObjectFactory();
      GenericObjectPoolConfig<Recognition> config =
          new GenericObjectPoolConfig<>();
      config.setMaxTotal(objectPoolSize);
      config.setMaxIdle(objectPoolSize);
      config.setMinIdle(objectPoolSize);
      recognitionGenericObjectPool =
          new GenericObjectPool<>(recognitionObjectFactory, config);
    }
    lock.unlock();
    return recognitionGenericObjectPool;
  }
}

class RealtimeRecognizeTask implements Runnable {
  private static final Object lock = new Object();
  private Path[] filePaths;

  public RealtimeRecognizeTask(Path[] filePaths) {
    this.filePaths = filePaths;
  }

  /**
   * Set your DashScope API key. In
   * fact, if you have set DASHSCOPE_API_KEY in your environment variable, you
   * can ignore this, and the SDK will automatically get the api_key from the
   * environment variable
   */
  private static String getDashScopeApiKey() throws NoApiKeyException {
    String dashScopeApiKey = null;
    try {
      ApiKey apiKey = new ApiKey();
      dashScopeApiKey =
          ApiKey.getApiKey(null); // Retrieve from environment variable.
    } catch (NoApiKeyException e) {
      System.out.println("No API key found in environment.");
    }
    if (dashScopeApiKey == null) {
      // If you cannot set api_key in your environment variable,
      // you can set it here by code
      dashScopeApiKey = "your-dashscope-apikey";
    }
    return dashScopeApiKey;
  }

  public void runCallback() {
    for (Path filePath : filePaths) {
      // Create recognition params
      // you can customize the recognition parameters, like model, format,
      // sample_rate
      RecognitionParam param = null;
      try {
        param =
            RecognitionParam.builder()
                .model("paraformer-realtime-v2")
                .format(
                    "pcm") // 'pcm'、'wav'、'opus'、'speex'、'aac'、'amr', you
                // can check the supported formats in the document
                .sampleRate(16000) // supported 8000、16000
                .apiKey(getDashScopeApiKey()) // use getDashScopeApiKey to get
                // api key.
                .build();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }

      Recognition recognizer = null;
      // if recv onError
      final boolean[] hasError = {false};
      try {
        recognizer = RecognitionObjectPool.getInstance().borrowObject();

        String threadName = Thread.currentThread().getName();

        ResultCallback<RecognitionResult> callback =
            new ResultCallback<RecognitionResult>() {
              @Override
              public void onEvent(RecognitionResult message) {
                synchronized (lock) {
                  if (message.isSentenceEnd()) {
                    System.out.println("[process " + threadName
                        + "] Fix:" + message.getSentence().getText());
                  } else {
                    System.out.println("[process " + threadName
                        + "] Result: " + message.getSentence().getText());
                  }
                }
              }

              @Override
              public void onComplete() {
                System.out.println("[" + threadName + "] Recognition complete");
              }

              @Override
              public void onError(Exception e) {
                System.out.println("[" + threadName
                    + "] RecognitionCallback error: " + e.getMessage());
                hasError[0] = true;
              }
            };
        // Please replace the path with your audio file path
        System.out.println(
            "[" + threadName + "] Input file_path is: " + filePath);
        FileInputStream fis = null;
        // Read file and send audio by chunks
        try {
          fis = new FileInputStream(filePath.toFile());
        } catch (Exception e) {
          System.out.println("Error when loading file: " + filePath);
          e.printStackTrace();
        }
        // set param & callback
        recognizer.call(param, callback);

        // chunk size set to 100 ms for 16KHz sample rate
        byte[] buffer = new byte[3200];
        int bytesRead;
        // Loop to read chunks of the file
        while ((bytesRead = fis.read(buffer)) != -1) {
          ByteBuffer byteBuffer;
          if (bytesRead < buffer.length) {
            byteBuffer = ByteBuffer.wrap(buffer, 0, bytesRead);
          } else {
            byteBuffer = ByteBuffer.wrap(buffer);
          }
          // Send the ByteBuffer to the recognition instance
          recognizer.sendAudioFrame(byteBuffer);
          Thread.sleep(100);
          buffer = new byte[3200];
        }
        System.out.println(
            "[" + threadName + "] send audio done");
        recognizer.stop();
        System.out.println(
            "[" + threadName + "] asr task finished");
      } catch (Exception e) {
        e.printStackTrace();
        hasError[0] = true;
      }
      if (recognizer != null) {
        try {
          if (hasError[0] == true) {
            // Invalid the recognition object error.
            recognizer.getDuplexApi().close(1000, "bye");
            RecognitionObjectPool.getInstance().invalidateObject(recognizer);
          } else {
            // Return the recognition object to the pool if no error or exception.
            RecognitionObjectPool.getInstance().returnObject(recognizer);
          }
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
    }
  }

  @Override
  public void run() {
    runCallback();
  }
}

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