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

Paraformer实时语音识别内部通过WebSocket协议实现，在高并发场景下，WebSocket连接可能会被不断地创建，从而带来较大的资源消耗。

在使用DashScope Java SDK时，您可以根据服务器的实际情况，通过给连接池、对象池设定合理的大小来降低运行开销。

前提条件

已开通服务并获取API Key

推荐配置

连接池和对象池不是越多越好，过少或过多都会导致程序运行变慢。建议您根据自己服务器的实际规格进行配置。

我们在服务器上只运行Paraformer实时语音识别服务的情况下，进行测试后得到了如下推荐配置供您参考：

其中单机并发数指的是同一时刻正在运行的Paraformer实时语音识别任务数，也可以理解为工作线程数。

可配置参数

关于如何配置环境变量，可参考配置API Key到环境变量。

示例代码

下面为使用资源池的示例代码。其中，对象池为全局单例对象。

• 每个主账号默认每秒可提交20个Paraformer实时语音识别任务。

  如需开通更高QPS请联系我们。

• 在运行示例前，请提前下载好asr_example.wav示例音频，或替换为本地音频文件。

  asr_example.wav

• 您需要在项目中引入DashScope和org.apache.commons.pool2相关的包，DashScope要求版本号>=2.16.9。

  以Maven和Gradle为例，配置如下：

  Maven

  1. 打开您的Maven项目的pom.xml文件。

  2. 在<dependencies>标签内添加以下依赖信息。

  <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>dashscope-sdk-java</artifactId>
      <!-- 请将 'the-latest-version' 替换为2.16.9及以上版本，可在如下链接查询相关版本号：https://mvnrepository.com/artifact/com.alibaba/dashscope-sdk-java -->
      <version>the-latest-version</version>
  </dependency>
  
  <dependency>
      <groupId>org.apache.commons</groupId>
      <artifactId>commons-pool2</artifactId>
      <!-- 请将 'the-latest-version' 替换为最新版本，可在如下链接查询相关版本号：https://mvnrepository.com/artifact/org.apache.commons/commons-pool2 -->
      <version>the-latest-version</version>
  </dependency>

  3. 保存pom.xml文件。

  4. 使用Maven命令（如mvn clean install或mvn compile）来更新项目依赖

  Gradle

  1. 打开您的Gradle项目的build.gradle文件。

  2. 在dependencies块内添加以下依赖信息。

     dependencies {
         // 请将 'the-latest-version' 替换为2.16.9及以上版本，可在如下链接查询相关版本号：https://mvnrepository.com/artifact/com.alibaba/dashscope-sdk-java
         implementation group: 'com.alibaba', name: 'dashscope-sdk-java', version: 'the-latest-version'
         
         // 请将 'the-latest-version' 替换为最新版本，可在如下链接查询相关版本号：https://mvnrepository.com/artifact/org.apache.commons/commons-pool2
         implementation group: 'org.apache.commons', name: 'commons-pool2', version: 'the-latest-version'
     }

  3. 保存build.gradle文件。

  4. 在命令行中，切换到您的项目根目录，执行以下Gradle命令来更新项目依赖。

     ./gradlew build --refresh-dependencies

     或者，如果您使用的是Windows系统，命令应为：

     gradlew build --refresh-dependencies

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.time.LocalDateTime;
import java.util.concurrent.CountDownLatch;
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 RecognitionWithRealtimeApiConcurrently {
    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(1, 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. More information: <a
     * href="https://help.aliyun.com/document_detail/2712195.html">...</a> 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 for more information, please refer to
            // https://help.aliyun.com/document_detail/2712536.html
            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 (NoApiKeyException e) {
                throw new RuntimeException(e);
            }

            Recognition recognizer = null;

            try {
                recognizer = RecognitionObjectPool.getInstance().borrowObject();
                CountDownLatch latch = new CountDownLatch(1);

                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");
                                latch.countDown();
                            }

                            @Override
                            public void onError(Exception e) {
                                System.out.println("[" + threadName
                                        + "] RecognitionCallback error: " + e.getMessage());
                            }
                        };
                // set param & callback
                recognizer.call(param, callback);
                // Please replace the path with your audio file path
                System.out.println(
                        "[" + threadName + "] Input file_path is: " + filePath);
                // Read file and send audio by chunks
                try (FileInputStream fis = new FileInputStream(filePath.toFile())) {
                    // 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(LocalDateTime.now());
                } catch (Exception e) {
                    e.printStackTrace();
                    recognizer.getDuplexApi().close(1000, "bye");
                }

                recognizer.stop();
                // wait for the recognition to complete
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            } catch (Exception e) {
                e.printStackTrace();
                recognizer.getDuplexApi().close(1000, "bye");
            } finally {
                if (recognizer != null) {
                    try {
                        // Return the recognition object to the pool
                        RecognitionObjectPool.getInstance().returnObject(recognizer);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

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

常见异常