高并发场景调用 DashScope Java SDK 最佳实践

import com.alibaba.dashscope.audio.tts.SpeechSynthesisAudioFormat;
import com.alibaba.dashscope.audio.tts.SpeechSynthesisParam;
import com.alibaba.dashscope.audio.tts.SpeechSynthesisResult;
import com.alibaba.dashscope.audio.tts.SpeechSynthesizer;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.dashscope.exception.NoApiKeyException;
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.util.Random;
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 TTS 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.
 */

@Slf4j
public class SynthesizeTextToSpeechUsingSambertConcurrently {
    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(SambertObjectPool.SAMBERT_OBJECTPOOL_SIZE_ENV, SambertObjectPool.DEFAULT_CONNECTION_POOL_SIZE);
        checkoutEnv("DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST", 32);

        // Record task start time
        int runTimes = 1;

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

        for (int i = 0; i < runTimes; i++) {
            executorService.submit(new SynthesizeTask(new String[]{
                    "床前明月光，",
                    "疑似地上霜。",
                    "举头望明月，",
                    "低头思故乡。"
            }));
        }

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

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 SambertObjectPool {
    public static GenericObjectPool<SpeechSynthesizer> synthesizerPool;
    public static String SAMBERT_OBJECTPOOL_SIZE_ENV = "SAMBERT_OBJECTPOOL_SIZE";
    public static int DEFAULT_CONNECTION_POOL_SIZE = 500;
    private static Lock lock = new java.util.concurrent.locks.ReentrantLock();
    public static int getObjectivePoolSize() {
        try {
            Integer n = Integer.parseInt(System.getenv(SAMBERT_OBJECTPOOL_SIZE_ENV));
            return n;
        } catch (NumberFormatException e) {
            return DEFAULT_CONNECTION_POOL_SIZE;
        }
    }
    public static GenericObjectPool<SpeechSynthesizer> getInstance() {
        lock.lock();
        if (synthesizerPool == null) {
            // You can set the object pool size here. or in environment variable
            // SAMBERT_OBJECTPOOL_SIZE It is recommended to set it to 1.5 to 2 times
            // your server's maximum concurrent connections.
            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 SynthesizeTask implements Runnable {
    String[] textList;
    String requestId;
    long timeCost;
    public SynthesizeTask(String[] textList) {
        this.textList = textList;
    }
    @Override
    public void run() {
        // sleep random time before start task, avoid creating too much websocket at the same time.
        Random random = new Random();
        try {
            Thread.sleep(random.nextInt(30*1000));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        for (String text:textList) {
            SpeechSynthesizer synthesizer = null;
            long startTime = System.currentTimeMillis();

            try {
                CountDownLatch latch = new CountDownLatch(1);
                class ReactCallback extends ResultCallback<SpeechSynthesisResult> {
                    ReactCallback() {}

                    @Override
                    public void onEvent(SpeechSynthesisResult message) {
                        if (message.getAudioFrame() != null) {
                            try {
                                byte[] bytesArray = message.getAudioFrame().array();
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }

                    @Override
                    public void onComplete() {
                        latch.countDown();
                    }

                    @Override
                    public void onError(Exception e) {
                        System.out.println(e.getMessage());
                        e.printStackTrace();
                        latch.countDown();
                    }
                }

                // you can set your dashscope apikey here by code or in environment
                // variable DASHSCOPE_API_KEY
                String dashScopeApiKey = System.getenv("DASHSCOPE_API_KEY");

                SpeechSynthesisParam param =
                        SpeechSynthesisParam.builder()
                                .model("sambert-zhichu-v1")
                                .format(SpeechSynthesisAudioFormat.MP3) // 使用PCM或者MP3
                                .text(text)
                                .enablePhonemeTimestamp(true)
                                .enableWordTimestamp(true)
                                .apiKey(dashScopeApiKey)
                                .build();

                try {
                    synthesizer = SambertObjectPool.getInstance().borrowObject();
                    synthesizer.call(param, new ReactCallback());
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    requestId = synthesizer.getLastRequestId();
                } catch (Exception e) {
                    System.out.println("Exception e: " + e.toString());
                    synthesizer.getSyncApi().close(1000, "bye");
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                if (synthesizer != null) {
                    try {
                        // Return the SpeechSynthesizer object to the pool
                        SambertObjectPool.getInstance().returnObject(synthesizer);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
            long endTime = System.currentTimeMillis();
            timeCost = endTime - startTime;
            System.out.println("[线程" + Thread.currentThread() + "] 语音合成任务:(" + text + ")结束。耗时" + timeCost + "ms, RequestId" + requestId);
        }
    }
}

// 将下面这段代码放在try-catch块中
synthesizer.getSyncApi().close(1000, "bye");