语音合成高并发最佳实践

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.utils.Constants;
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.FileOutputStream;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

public class PooledSambert {
    static class Factory extends BasePooledObjectFactory<SpeechSynthesizer> {
        @Override
        public SpeechSynthesizer create() { return new SpeechSynthesizer(); }

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

        @Override
        public void destroyObject(PooledObject<SpeechSynthesizer> value) {
            value.getObject().getSyncApi().close(1000, "done");
        }
    }

    static void synthesize(GenericObjectPool<SpeechSynthesizer> pool, int index, String text)
            throws Exception {
        SpeechSynthesizer synthesizer = pool.borrowObject();
        boolean successful = false;
        CountDownLatch finished = new CountDownLatch(1);
        AtomicReference<Exception> error = new AtomicReference<>();
        try (FileOutputStream audio = new FileOutputStream("sambert_" + index + ".mp3")) {
            SpeechSynthesisParam param = SpeechSynthesisParam.builder()
                .apiKey(System.getenv("DASHSCOPE_API_KEY"))
                .model("sambert-zhichu-v1")
                .text(text)
                .format(SpeechSynthesisAudioFormat.MP3)
                .enablePhonemeTimestamp(true)
                .enableWordTimestamp(true)
                .build();
            synthesizer.call(param, new ResultCallback<SpeechSynthesisResult>() {
                @Override
                public void onEvent(SpeechSynthesisResult result) {
                    if (result.getAudioFrame() != null) {
                        byte[] bytes = result.getAudioFrame().array();
                        try { audio.write(bytes); }
                        catch (Exception e) { error.compareAndSet(null, e); }
                    }
                }

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

                @Override
                public void onError(Exception e) {
                    error.compareAndSet(null, e);
                    finished.countDown();
                }
            });
            if (!finished.await(60, TimeUnit.SECONDS)) {
                throw new TimeoutException("Synthesis timed out");
            }
            if (error.get() != null) { throw error.get(); }
            successful = true;
        } finally {
            if (successful) { pool.returnObject(synthesizer); }
            else { pool.invalidateObject(synthesizer); }
        }
    }

    public static void main(String[] args) throws Exception {
        Constants.baseWebsocketApiUrl = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference";
        int poolSize = Integer.parseInt(System.getenv().getOrDefault("SAMBERT_OBJECTPOOL_SIZE", "500"));
        GenericObjectPoolConfig<SpeechSynthesizer> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(poolSize);
        config.setMaxIdle(poolSize);
        config.setMaxWait(Duration.ofSeconds(60));
        ExecutorService executor = Executors.newFixedThreadPool(3);
        try (GenericObjectPool<SpeechSynthesizer> pool = new GenericObjectPool<>(new Factory(), config)) {
            String[] texts = {"床前明月光，疑似地上霜。", "举头望明月，低头思故乡。", "你好，欢迎使用阿里云百炼。"};
            List<Future<?>> futures = new ArrayList<>();
            for (int i = 0; i < texts.length; i++) {
                final int index = i;
                futures.add(executor.submit(() -> {
                    try { synthesize(pool, index, texts[index]); }
                    catch (Exception e) { throw new RuntimeException(e); }
                }));
            }
            Exception failure = null;
            for (Future<?> future : futures) {
                try { future.get(); }
                catch (ExecutionException e) { failure = e; }
            }
            if (failure != null) { throw failure; }
        } finally {
            executor.shutdownNow();
        }
    }
}