Alibaba Cloud Model Studio：即時語音合成

Go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	// 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
	wsURL      = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"
	outputFile = "output.mp3"
)

func main() {
	// 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
	// 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：apiKey := "sk-xxx"
	apiKey := os.Getenv("DASHSCOPE_API_KEY")

	// 清空輸出檔案
	os.Remove(outputFile)
	os.Create(outputFile)

	// 串連WebSocket
	header := make(http.Header)
	header.Add("X-DashScope-DataInspection", "enable")
	header.Add("Authorization", fmt.Sprintf("bearer %s", apiKey))

	conn, resp, err := websocket.DefaultDialer.Dial(wsURL, header)
	if err != nil {
		if resp != nil {
			fmt.Printf("串連失敗 HTTP狀態代碼: %d\n", resp.StatusCode)
		}
		fmt.Println("串連失敗:", err)
		return
	}
	defer conn.Close()

	// 產生任務ID
	taskID := uuid.New().String()
	fmt.Printf("產生任務ID: %s\n", taskID)

	// 發送run-task事件
	runTaskCmd := map[string]interface{}{
		"header": map[string]interface{}{
			"action":    "run-task",
			"task_id":   taskID,
			"streaming": "duplex",
		},
		"payload": map[string]interface{}{
			"task_group": "audio",
			"task":       "tts",
			"function":   "SpeechSynthesizer",
			"model":      "qwen-audio-3.0-tts-flash",
			"parameters": map[string]interface{}{
				"text_type":   "PlainText",
				"voice":       "longanhuan_v3.6",
				"format":      "mp3",
				"sample_rate": 22050,
				"volume":      50,
				"rate":        1,
				"pitch":       1,
				// 如果enable_ssml設為true，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
				"enable_ssml": false,
			},
			"input": map[string]interface{}{},
		},
	}

	runTaskJSON, _ := json.Marshal(runTaskCmd)
	fmt.Printf("發送run-task事件: %s\n", string(runTaskJSON))

	err = conn.WriteMessage(websocket.TextMessage, runTaskJSON)
	if err != nil {
		fmt.Println("發送run-task失敗:", err)
		return
	}

	textSent := false

	// 處理訊息
	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			fmt.Println("讀取訊息失敗:", err)
			break
		}

		// 處理二進位訊息
		if messageType == websocket.BinaryMessage {
			fmt.Printf("收到二進位訊息，長度: %d\n", len(message))
			file, _ := os.OpenFile(outputFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0644)
			file.Write(message)
			file.Close()
			continue
		}

		// 處理簡訊
		messageStr := string(message)
		fmt.Printf("收到簡訊: %s\n", strings.ReplaceAll(messageStr, "\n", ""))

		// 簡單解析JSON擷取event類型
		var msgMap map[string]interface{}
		if json.Unmarshal(message, &msgMap) == nil {
			if header, ok := msgMap["header"].(map[string]interface{}); ok {
				if event, ok := header["event"].(string); ok {
					fmt.Printf("事件類型: %s\n", event)

					switch event {
					case "task-started":
						fmt.Println("=== 收到task-started事件 ===")

						if !textSent {
							// 發送continue-task事件

							texts := []string{"床前明月光，疑是地上霜。", "舉頭望明月，低頭思故鄉。"}

							for _, text := range texts {
								continueTaskCmd := map[string]interface{}{
									"header": map[string]interface{}{
										"action":    "continue-task",
										"task_id":   taskID,
										"streaming": "duplex",
									},
									"payload": map[string]interface{}{
										"input": map[string]interface{}{
											"text": text,
										},
									},
								}

								continueTaskJSON, _ := json.Marshal(continueTaskCmd)
								fmt.Printf("發送continue-task事件: %s\n", string(continueTaskJSON))

								err = conn.WriteMessage(websocket.TextMessage, continueTaskJSON)
								if err != nil {
									fmt.Println("發送continue-task失敗:", err)
									return
								}
							}

							textSent = true

							// 延遲發送finish-task
							time.Sleep(500 * time.Millisecond)

							// 發送finish-task事件
							finishTaskCmd := map[string]interface{}{
								"header": map[string]interface{}{
									"action":    "finish-task",
									"task_id":   taskID,
									"streaming": "duplex",
								},
								"payload": map[string]interface{}{
									"input": map[string]interface{}{},
								},
							}

							finishTaskJSON, _ := json.Marshal(finishTaskCmd)
							fmt.Printf("發送finish-task事件: %s\n", string(finishTaskJSON))

							err = conn.WriteMessage(websocket.TextMessage, finishTaskJSON)
							if err != nil {
								fmt.Println("發送finish-task失敗:", err)
								return
							}
						}

					case "task-finished":
						fmt.Println("=== 任務完成 ===")
						return

					case "task-failed":
						fmt.Println("=== 任務失敗 ===")
						if header["error_message"] != nil {
							fmt.Printf("錯誤資訊: %s\n", header["error_message"])
						}
						return

					case "result-generated":
						fmt.Println("收到result-generated事件")
					}
				}
			}
		}
	}
}

C#

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

class Program {
    // 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    // 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：private static readonly string ApiKey = "sk-xxx"
    private static readonly string ApiKey = Environment.GetEnvironmentVariable("DASHSCOPE_API_KEY") ?? throw new InvalidOperationException("DASHSCOPE_API_KEY environment variable is not set.");

    // 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
    private const string WebSocketUrl = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference";
    // 輸出檔案路徑
    private const string OutputFilePath = "output.mp3";

    // WebSocket用戶端
    private static ClientWebSocket _webSocket = new ClientWebSocket();
    // 取消令牌源
    private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    // 任務ID
    private static string? _taskId;
    // 任務是否已啟動
    private static TaskCompletionSource<bool> _taskStartedTcs = new TaskCompletionSource<bool>();

    static async Task Main(string[] args) {
        try {
            // 清空輸出檔案
            ClearOutputFile(OutputFilePath);

            // 串連WebSocket服務
            await ConnectToWebSocketAsync(WebSocketUrl);

            // 啟動接收訊息的任務
            Task receiveTask = ReceiveMessagesAsync();

            // 發送run-task事件
            _taskId = GenerateTaskId();
            await SendRunTaskCommandAsync(_taskId);

            // 等待task-started事件
            await _taskStartedTcs.Task;

            // 持續發送continue-task事件
            string[] texts = {
                "床前明月光",
                "疑是地上霜",
                "舉頭望明月",
                "低頭思故鄉"
            };
            foreach (string text in texts) {
                await SendContinueTaskCommandAsync(text);
            }

            // 發送finish-task事件
            await SendFinishTaskCommandAsync(_taskId);

            // 等待接收任務完成
            await receiveTask;

            Console.WriteLine("任務完成，串連已關閉。");
        } catch (OperationCanceledException) {
            Console.WriteLine("任務被取消。");
        } catch (Exception ex) {
            Console.WriteLine($"發生錯誤：{ex.Message}");
        } finally {
            _cancellationTokenSource.Cancel();
            _webSocket.Dispose();
        }
    }

    private static void ClearOutputFile(string filePath) {
        if (File.Exists(filePath)) {
            File.WriteAllText(filePath, string.Empty);
            Console.WriteLine("輸出檔案已清空。");
        } else {
            Console.WriteLine("輸出檔案不存在，無需清空。");
        }
    }

    private static async Task ConnectToWebSocketAsync(string url) {
        var uri = new Uri(url);
        if (_webSocket.State == WebSocketState.Connecting || _webSocket.State == WebSocketState.Open) {
            return;
        }

        // 設定WebSocket串連的頭部資訊
        _webSocket.Options.SetRequestHeader("Authorization", $"bearer {ApiKey}");
        _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");

        try {
            await _webSocket.ConnectAsync(uri, _cancellationTokenSource.Token);
            Console.WriteLine("已成功串連到WebSocket服務。");
        } catch (OperationCanceledException) {
            Console.WriteLine("WebSocket串連被取消。");
        } catch (Exception ex) {
            Console.WriteLine($"WebSocket串連失敗: {ex.Message}");
            throw;
        }
    }

    private static async Task SendRunTaskCommandAsync(string taskId) {
        var command = CreateCommand("run-task", taskId, "duplex", new {
            task_group = "audio",
            task = "tts",
            function = "SpeechSynthesizer",
            model = "qwen-audio-3.0-tts-flash",
            parameters = new
            {
                text_type = "PlainText",
                voice = "longanhuan_v3.6",
                format = "mp3",
                sample_rate = 22050,
                volume = 50,
                rate = 1,
                pitch = 1,
                // 如果enable_ssml設為true，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
                enable_ssml = false
            },
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("已發送run-task事件。");
    }

    private static async Task SendContinueTaskCommandAsync(string text) {
        if (_taskId == null) {
            throw new InvalidOperationException("任務ID未初始化。");
        }

        var command = CreateCommand("continue-task", _taskId, "duplex", new {
            input = new {
                text
            }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("已發送continue-task事件。");
    }

    private static async Task SendFinishTaskCommandAsync(string taskId) {
        var command = CreateCommand("finish-task", taskId, "duplex", new {
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("已發送finish-task事件。");
    }

    private static async Task SendJsonMessageAsync(string message) {
        var buffer = Encoding.UTF8.GetBytes(message);
        try {
            await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, _cancellationTokenSource.Token);
        } catch (OperationCanceledException) {
            Console.WriteLine("訊息發送被取消。");
        }
    }

    private static async Task ReceiveMessagesAsync() {
        while (_webSocket.State == WebSocketState.Open) {
            var response = await ReceiveMessageAsync();
            if (response != null) {
                var eventStr = response.RootElement.GetProperty("header").GetProperty("event").GetString();
                switch (eventStr) {
                    case "task-started":
                        Console.WriteLine("任務已啟動。");
                        _taskStartedTcs.TrySetResult(true);
                        break;
                    case "task-finished":
                        Console.WriteLine("任務已完成。");
                        _cancellationTokenSource.Cancel();
                        break;
                    case "task-failed":
                        Console.WriteLine("任務失敗：" + response.RootElement.GetProperty("header").GetProperty("error_message").GetString());
                        _cancellationTokenSource.Cancel();
                        break;
                    default:
                        // result-generated可在此處理
                        break;
                }
            }
        }
    }

    private static async Task<JsonDocument?> ReceiveMessageAsync() {
        var buffer = new byte[1024 * 4];
        var segment = new ArraySegment<byte>(buffer);

        try {
            WebSocketReceiveResult result = await _webSocket.ReceiveAsync(segment, _cancellationTokenSource.Token);

            if (result.MessageType == WebSocketMessageType.Close) {
                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", _cancellationTokenSource.Token);
                return null;
            }

            if (result.MessageType == WebSocketMessageType.Binary) {
                // 處理位元據
                Console.WriteLine("接收到位元據...");

                // 將位元據儲存到檔案
                using (var fileStream = new FileStream(OutputFilePath, FileMode.Append)) {
                    fileStream.Write(buffer, 0, result.Count);
                }

                return null;
            }

            string message = Encoding.UTF8.GetString(buffer, 0, result.Count);
            return JsonDocument.Parse(message);
        } catch (OperationCanceledException) {
            Console.WriteLine("訊息接收被取消。");
            return null;
        }
    }

    private static string GenerateTaskId() {
        return Guid.NewGuid().ToString("N").Substring(0, 32);
    }

    private static string CreateCommand(string action, string taskId, string streaming, object payload) {
        var command = new {
            header = new {
                action,
                task_id = taskId,
                streaming
            },
            payload
        };

        return JsonSerializer.Serialize(command);
    }
}

PHP

範例程式碼目錄結構為：

my-php-project/

├── composer.json

├── vendor/

└── index.php

composer.json內容如下，相關依賴的版本號碼請根據實際情況自行決定：

{
    "require": {
        "react/event-loop": "^1.3",
        "react/socket": "^1.11",
        "react/stream": "^1.2",
        "react/http": "^1.1",
        "ratchet/pawl": "^0.4"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    }
}

index.php內容如下：

<?php

require __DIR__ . '/vendor/autoload.php';

use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;

// 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
// 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：$api_key = "sk-xxx"
$api_key = getenv("DASHSCOPE_API_KEY");
// 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
$websocket_url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'; // WebSocket伺服器位址
$output_file = 'output.mp3'; // 輸出檔案路徑

$loop = Loop::get();

if (file_exists($output_file)) {
    // 清空檔案內容
    file_put_contents($output_file, '');
}

// 建立自訂的連接器
$socketConnector = new SocketConnector($loop, [
    'tcp' => [
        'bindto' => '0.0.0.0:0',
    ],
    // 警告：關閉 TLS 認證校正存在中間人攻擊風險，僅限本地調試；生產環境務必將 verify_peer/verify_peer_name 設為 true
    'tls' => [
        'verify_peer' => false,
        'verify_peer_name' => false,
    ],
]);

$connector = new Connector($loop, $socketConnector);

$headers = [
    'Authorization' => 'bearer ' . $api_key,
    'X-DashScope-DataInspection' => 'enable'
];

$connector($websocket_url, [], $headers)->then(function ($conn) use ($loop, $output_file) {
    echo "串連到WebSocket伺服器\n";

    // 產生任務ID
    $taskId = generateTaskId();

    // 發送 run-task 事件
    sendRunTaskMessage($conn, $taskId);

    // 定義發送 continue-task 事件的函數
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
        // 待發送的文本
        $texts = ["床前明月光", "疑是地上霜", "舉頭望明月", "低頭思故鄉"];
        $continueTaskCount = 0;
        foreach ($texts as $text) {
            $continueTaskMessage = json_encode([
                "header" => [
                    "action" => "continue-task",
                    "task_id" => $taskId,
                    "streaming" => "duplex"
                ],
                "payload" => [
                    "input" => [
                        "text" => $text
                    ]
                ]
            ]);
            echo "準備發送continue-task事件: " . $continueTaskMessage . "\n";
            $conn->send($continueTaskMessage);
            $continueTaskCount++;
        }
        echo "發送的continue-task事件個數為：" . $continueTaskCount . "\n";

        // 發送 finish-task 事件
        sendFinishTaskMessage($conn, $taskId);
    };

    // 標記是否收到 task-started 事件
    $taskStarted = false;

    // 監聽訊息
    $conn->on('message', function($msg) use ($conn, $sendContinueTask, $loop, &$taskStarted, $taskId, $output_file) {
        if ($msg->isBinary()) {
            // 寫入位元據到本地檔案
            file_put_contents($output_file, $msg->getPayload(), FILE_APPEND);
        } else {
            // 處理非二進位訊息
            $response = json_decode($msg, true);

            if (isset($response['header']['event'])) {
                handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, $taskStarted);
            } else {
                echo "未知的訊息格式\n";
            }
        }
    });

    // 監聽串連關閉
    $conn->on('close', function($code = null, $reason = null) {
        echo "串連已關閉\n";
        if ($code !== null) {
            echo "關閉代碼: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "關閉原因：" . $reason . "\n";
        }
    });
}, function ($e) {
    echo "無法串連：{$e->getMessage()}\n";
});

$loop->run();

/**
 * 產生任務ID
 * @return string
 */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}

/**
 * 發送 run-task 事件
 * @param $conn
 * @param $taskId
 */
function sendRunTaskMessage($conn, $taskId) {
    $runTaskMessage = json_encode([
        "header" => [
            "action" => "run-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "task_group" => "audio",
            "task" => "tts",
            "function" => "SpeechSynthesizer",
            "model" => "qwen-audio-3.0-tts-flash",
            "parameters" => [
                "text_type" => "PlainText",
                "voice" => "longanhuan_v3.6",
                "format" => "mp3",
                "sample_rate" => 22050,
                "volume" => 50,
                "rate" => 1,
                "pitch" => 1,
                // 如果enable_ssml設為true，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
                "enable_ssml" => false
            ],
            "input" => (object) []
        ]
    ]);
    echo "準備發送run-task事件: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task事件已發送\n";
}

/**
 * 讀取音頻檔案
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "無法讀取音頻檔案\n";
    }
    return $voiceData;
}

/**
 * 分割音頻資料
 * @param string $data
 * @param int $chunkSize
 * @return array
 */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}

/**
 * 發送 finish-task 事件
 * @param $conn
 * @param $taskId
 */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
        "header" => [
            "action" => "finish-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "input" => (object) []
        ]
    ]);
    echo "準備發送finish-task事件: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task事件已發送\n";
}

/**
 * 處理事件
 * @param $conn
 * @param $response
 * @param $sendContinueTask
 * @param $loop
 * @param $taskId
 * @param $taskStarted
 */
function handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, &$taskStarted) {
    switch ($response['header']['event']) {
        case 'task-started':
            echo "任務開始，發送continue-task事件...\n";
            $taskStarted = true;
            // 發送 continue-task 事件
            $sendContinueTask();
            break;
        case 'result-generated':
            // 收到result-generated事件
            break;
        case 'task-finished':
            echo "任務完成\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "任務失敗\n";
            echo "錯誤碼：" . $response['header']['error_code'] . "\n";
            echo "錯誤資訊：" . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "錯誤：" . $response['payload']['message'] . "\n";
            break;
        default:
            echo "未知事件：" . $response['header']['event'] . "\n";
            break;
    }

    // 如果任務已完成，關閉串連
    if ($response['header']['event'] == 'task-finished') {
        // 等待1秒以確保所有資料都已傳輸完畢
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "用戶端關閉串連\n";
        });
    }

    // 如果沒有收到 task-started 事件，關閉串連
    if (!$taskStarted && in_array($response['header']['event'], ['task-failed', 'error'])) {
        $conn->close();
    }
}

Node.js

需安裝相關依賴：

npm install ws
npm install uuid

範例程式碼如下：

const WebSocket = require('ws');
const fs = require('fs');
const uuid = require('uuid').v4;

// 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
// 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：const apiKey = "sk-xxx"
const apiKey = process.env.DASHSCOPE_API_KEY;
// 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
const url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference';
// 輸出檔案路徑
const outputFilePath = 'output.mp3';

// 清空輸出檔案
fs.writeFileSync(outputFilePath, '');

// 建立WebSocket用戶端
const ws = new WebSocket(url, {
  headers: {
    Authorization: `bearer ${apiKey}`,
    'X-DashScope-DataInspection': 'enable'
  }
});

let taskStarted = false;
let taskId = uuid();

ws.on('open', () => {
  console.log('已串連到WebSocket伺服器');

  // 發送run-task事件
  const runTaskMessage = JSON.stringify({
    header: {
      action: 'run-task',
      task_id: taskId,
      streaming: 'duplex'
    },
    payload: {
      task_group: 'audio',
      task: 'tts',
      function: 'SpeechSynthesizer',
      model: 'qwen-audio-3.0-tts-flash',
      parameters: {
        text_type: 'PlainText',
        voice: 'longanhuan_v3.6', // 音色
        format: 'mp3', // 音頻格式
        sample_rate: 22050, // 採樣率
        volume: 50, // 音量
        rate: 1, // 語速
        pitch: 1, // 音調
        enable_ssml: false // 是否開啟SSML功能。如果enable_ssml設為true，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
      },
      input: {}
    }
  });
  ws.send(runTaskMessage);
  console.log('已發送run-task訊息');
});

const fileStream = fs.createWriteStream(outputFilePath, { flags: 'a' });
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // 寫入位元據到檔案
    fileStream.write(data);
  } else {
    const message = JSON.parse(data);

    switch (message.header.event) {
      case 'task-started':
        taskStarted = true;
        console.log('任務已開始');
        // 發送continue-task事件
        sendContinueTasks(ws);
        break;
      case 'task-finished':
        console.log('任務已完成');
        ws.close();
        fileStream.end(() => {
          console.log('檔案流已關閉');
        });
        break;
      case 'task-failed':
        console.error('任務失敗：', message.header.error_message);
        ws.close();
        fileStream.end(() => {
          console.log('檔案流已關閉');
        });
        break;
      default:
        // 可以在這裡處理result-generated
        break;
    }
  }
});

function sendContinueTasks(ws) {
  const texts = [
    '床前明月光，',
    '疑是地上霜。',
    '舉頭望明月，',
    '低頭思故鄉。'
  ];

  texts.forEach((text, index) => {
    setTimeout(() => {
      if (taskStarted) {
        const continueTaskMessage = JSON.stringify({
          header: {
            action: 'continue-task',
            task_id: taskId,
            streaming: 'duplex'
          },
          payload: {
            input: {
              text: text
            }
          }
        });
        ws.send(continueTaskMessage);
        console.log(`已發送continue-task，文本：${text}`);
      }
    }, index * 1000); // 每隔1秒發送一次
  });

  // 發送finish-task事件
  setTimeout(() => {
    if (taskStarted) {
      const finishTaskMessage = JSON.stringify({
        header: {
          action: 'finish-task',
          task_id: taskId,
          streaming: 'duplex'
        },
        payload: {
          input: {}
        }
      });
      ws.send(finishTaskMessage);
      console.log('已發送finish-task');
    }
  }, texts.length * 1000 + 1000); // 在所有continue-task事件發送完畢後1秒發送
}

ws.on('close', () => {
  console.log('已斷開與WebSocket伺服器的串連');
});

Java

建議使用 Java DashScope SDK 進行開發，請參見Java SDK。

以下是 Java WebSocket 直連樣本，運行前請匯入以下依賴：

• Java-WebSocket
• jackson-databind

推薦使用Maven或Gradle管理依賴包，其配置如下：

<dependencies>
    <!-- WebSocket Client -->
    <dependency>
        <groupId>org.java-websocket</groupId>
        <artifactId>Java-WebSocket</artifactId>
        <version>1.5.3</version>
    </dependency>

    <!-- JSON Processing -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.13.0</version>
    </dependency>
</dependencies>

// 省略其它代碼
dependencies {
  // WebSocket Client
  implementation 'org.java-websocket:Java-WebSocket:1.5.3'
  // JSON Processing
  implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.0'
}
// 省略其它代碼

Java代碼如下：

import com.fasterxml.jackson.databind.ObjectMapper;

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.*;

public class TTSWebSocketClient extends WebSocketClient {
    private final String taskId = UUID.randomUUID().toString();
    private final String outputFile = "output_" + System.currentTimeMillis() + ".mp3";
    private boolean taskFinished = false;

    public TTSWebSocketClient(URI serverUri, Map<String, String> headers) {
        super(serverUri, headers);
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        System.out.println("串連成功");

        // 發送run-task事件
        // 如果enable_ssml設為true，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
        String runTaskCommand = "{ \"header\": { \"action\": \"run-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"task_group\": \"audio\", \"task\": \"tts\", \"function\": \"SpeechSynthesizer\", \"model\": \"qwen-audio-3.0-tts-flash\", \"parameters\": { \"text_type\": \"PlainText\", \"voice\": \"longanhuan_v3.6\", \"format\": \"mp3\", \"sample_rate\": 22050, \"volume\": 50, \"rate\": 1, \"pitch\": 1, \"enable_ssml\": false }, \"input\": {} }}";
        send(runTaskCommand);
    }

    @Override
    public void onMessage(String message) {
        System.out.println("收到服務端返回的訊息：" + message);
        try {
            // Parse JSON message
            Map<String, Object> messageMap = new ObjectMapper().readValue(message, Map.class);

            if (messageMap.containsKey("header")) {
                Map<String, Object> header = (Map<String, Object>) messageMap.get("header");

                if (header.containsKey("event")) {
                    String event = (String) header.get("event");

                    if ("task-started".equals(event)) {
                        System.out.println("收到服務端返回的task-started事件");

                        List<String> texts = Arrays.asList(
                                "床前明月光，疑是地上霜",
                                "舉頭望明月，低頭思故鄉"
                        );

                        for (String text : texts) {
                            // 發送continue-task事件
                            sendContinueTask(text);
                        }

                        // 發送finish-task事件
                        sendFinishTask();
                    } else if ("task-finished".equals(event)) {
                        System.out.println("收到服務端返回的task-finished事件");
                        taskFinished = true;
                        closeConnection();
                    } else if ("task-failed".equals(event)) {
                        System.out.println("任務失敗：" + message);
                        closeConnection();
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("出現異常：" + e.getMessage());
        }
    }

    @Override
    public void onMessage(ByteBuffer message) {
        System.out.println("收到的二進位音頻資料大小為：" + message.remaining());

        try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
            byte[] buffer = new byte[message.remaining()];
            message.get(buffer);
            fos.write(buffer);
            System.out.println("音頻資料已寫入本地檔案" + outputFile + "中");
        } catch (IOException e) {
            System.err.println("音頻資料寫入本地檔案失敗：" + e.getMessage());
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("串連關閉：" + reason + " (" + code + ")");
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("報錯：" + ex.getMessage());
        ex.printStackTrace();
    }

    private void sendContinueTask(String text) {
        String command = "{ \"header\": { \"action\": \"continue-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": { \"text\": \"" + text + "\" } }}";
        send(command);
    }

    private void sendFinishTask() {
        String command = "{ \"header\": { \"action\": \"finish-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": {} }}";
        send(command);
    }

    private void closeConnection() {
        if (!isClosed()) {
            close();
        }
    }

    public static void main(String[] args) {
        try {
            // 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
            // 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：String apiKey = "sk-xxx"
            String apiKey = System.getenv("DASHSCOPE_API_KEY");
            if (apiKey == null || apiKey.isEmpty()) {
                System.err.println("請設定 DASHSCOPE_API_KEY 環境變數");
                return;
            }

            Map<String, String> headers = new HashMap<>();
            headers.put("Authorization", "bearer " + apiKey);
            // 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
            TTSWebSocketClient client = new TTSWebSocketClient(new URI("wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"), headers);

            client.connect();

            while (!client.isClosed() && !client.taskFinished) {
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            System.err.println("串連WebSocket服務失敗：" + e.getMessage());
            e.printStackTrace();
        }
    }
}

Python

建議使用 Python DashScope SDK 進行開發，請參見Python SDK。

以下是 Python WebSocket 直連樣本，運行前請匯入以下依賴：

pip uninstall websocket-client
pip uninstall websocket
pip install websocket-client

import websocket
import json
import uuid
import os
import time

class TTSClient:
    def __init__(self, api_key, uri):
        """
    初始化 TTSClient 執行個體

    參數:
        api_key (str): 鑒權用的 API Key
        uri (str): WebSocket 服務地址
    """
        self.api_key = api_key  # 替換為你的 API Key
        self.uri = uri  # 替換為你的 WebSocket 地址
        self.task_id = str(uuid.uuid4())  # 產生唯一任務 ID
        self.output_file = f"output_{int(time.time())}.mp3"  # 輸出音頻檔案路徑
        self.ws = None  # WebSocketApp 執行個體
        self.task_started = False  # 是否收到 task-started
        self.task_finished = False  # 是否收到 task-finished / task-failed

    def on_open(self, ws):
        """
    WebSocket 串連建立時回呼函數
    發送 run-task 事件開啟語音合成任務
    """
        print("WebSocket 已串連")

        # 構造 run-task 事件
        run_task_cmd = {
            "header": {
                "action": "run-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "task_group": "audio",
                "task": "tts",
                "function": "SpeechSynthesizer",
                "model": "qwen-audio-3.0-tts-flash",
                "parameters": {
                    "text_type": "PlainText",
                    "voice": "longanhuan_v3.6",
                    "format": "mp3",
                    "sample_rate": 22050,
                    "volume": 50,
                    "rate": 1,
                    "pitch": 1,
                    # 如果enable_ssml設為True，只允許發送一次continue-task事件，否則會報錯“Text request limit violated, expected 1.”
                    "enable_ssml": False
                },
                "input": {}
            }
        }

        # 發送 run-task 事件
        ws.send(json.dumps(run_task_cmd))
        print("已發送 run-task 事件")

    def on_message(self, ws, message):
        """
    接收到訊息時的回呼函數
    區分文本和二進位訊息處理
    """
        if isinstance(message, str):
            # 處理 JSON 簡訊
            try:
                msg_json = json.loads(message)
                print(f"收到 JSON 訊息: {msg_json}")

                if "header" in msg_json:
                    header = msg_json["header"]

                    if "event" in header:
                        event = header["event"]

                        if event == "task-started":
                            print("任務已啟動")
                            self.task_started = True

                            # 發送 continue-task 事件
                            texts = [
                                "床前明月光，疑是地上霜",
                                "舉頭望明月，低頭思故鄉"
                            ]

                            for text in texts:
                                self.send_continue_task(text)

                            # 所有 continue-task 發送完成後發送 finish-task
                            self.send_finish_task()

                        elif event == "task-finished":
                            print("任務已完成")
                            self.task_finished = True
                            self.close(ws)

                        elif event == "task-failed":
                            error_msg = msg_json.get("error_message", "未知錯誤")
                            print(f"任務失敗: {error_msg}")
                            self.task_finished = True
                            self.close(ws)

            except json.JSONDecodeError as e:
                print(f"JSON 解析失敗: {e}")
        else:
            # 處理二進位訊息（音頻資料）
            print(f"收到二進位訊息，大小: {len(message)} 位元組")
            with open(self.output_file, "ab") as f:
                f.write(message)
            print(f"已將音頻資料寫入本地檔案{self.output_file}中")

    def on_error(self, ws, error):
        """發生錯誤時的回調"""
        print(f"WebSocket 出錯: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        """串連關閉時的回調"""
        print(f"WebSocket 已關閉: {close_msg} ({close_status_code})")

    def send_continue_task(self, text):
        """發送 continue-task 事件，附帶要合成的常值內容"""
        cmd = {
            "header": {
                "action": "continue-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {
                    "text": text
                }
            }
        }

        self.ws.send(json.dumps(cmd))
        print(f"已發送 continue-task 事件，常值內容: {text}")

    def send_finish_task(self):
        """發送 finish-task 事件，結束語音合成任務"""
        cmd = {
            "header": {
                "action": "finish-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {}
            }
        }

        self.ws.send(json.dumps(cmd))
        print("已發送 finish-task 事件")

    def close(self, ws):
        """主動關閉串連"""
        if ws and ws.sock and ws.sock.connected:
            ws.close()
            print("已主動關閉串連")

    def run(self):
        """啟動 WebSocket 用戶端"""
        # 佈建要求頭部（鑒權）
        header = {
            "Authorization": f"bearer {self.api_key}",
            "X-DashScope-DataInspection": "enable"
        }

        # 建立 WebSocketApp 執行個體
        self.ws = websocket.WebSocketApp(
            self.uri,
            header=header,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )

        print("正在監聽 WebSocket 訊息...")
        self.ws.run_forever()  # 啟動長串連監聽

# 樣本使用方式
if __name__ == "__main__":
    # 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    # 若沒有配置環境變數，請用阿里雲百鍊API Key將下行替換為：API_KEY = "sk-xxx"
    API_KEY = os.environ.get("DASHSCOPE_API_KEY")
    # 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
    SERVER_URI = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"  # 替換為你的 WebSocket 地址

    client = TTSClient(API_KEY, SERVER_URI)
    client.run()

Python SDK

Python SDK 通過 SpeechSynthesizerObjectPool 管理和複用 SpeechSynthesizer 對象。

對象池在初始化時即建立指定數量的 SpeechSynthesizer 執行個體並建立 WebSocket 串連，擷取對象時可直接發起請求，降低首包延遲。歸還後串連保持活躍，等待下次複用。

實現步驟

1. 安裝依賴：安裝DashScope依賴（pip install -U dashscope）

2. 建立並設定物件池

   對象池大小推薦設為峰值並發數的 1.5~2 倍，且不應超過賬戶的 QPS 限制。

   建立全域單例對象池（初始化時建立串連，有一定耗時）：

from dashscope.audio.tts_v2 import SpeechSynthesizerObjectPool

connectionPool = SpeechSynthesizerObjectPool(max_size=20)
import dashscope
# 以下為華北2（北京）地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

3. 從對象池中擷取SpeechSynthesizer對象

   如果當前未歸還的對象數已超過池容量，系統會額外建立新對象。

   此類對象需重建立立串連，不具備複用效果。

speech_synthesizer = connectionPool.borrow_synthesizer(
    model='cosyvoice-v3-flash',
    voice='longanyang',
    seed=12382,
    callback=synthesizer_callback
)

4. 進行語音合成

   調用SpeechSynthesizer對象的call或streaming_call方法進行語音合成。

5. 歸還SpeechSynthesizer對象

   任務結束後歸還對象以供複用。

   不要歸還未完成任務或任務失敗的對象。

connectionPool.return_synthesizer(speech_synthesizer)

完整代碼

# !/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

def init_dashscope_api_key():
    '''
    Set your DashScope API-key. More information:
    https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
    '''
    # 新加坡地區和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-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 = '<your-dashscope-api-key>'  # set API-key manually

def synthesis_text_to_speech_and_play_by_streaming_mode(text, task_id):
    global USE_CONNECTION_POOL, connectionPool
    '''
    Synthesize speech with given text by streaming mode, async call and play the synthesized audio in real-time.
    for more information, please refer to https://www.alibabacloud.com/help/document_detail/2712523.html
    '''

    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):
            # print(f'recv speech synthsis message {message}')
            pass

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

    # Call the speech synthesizer callback
    synthesizer_callback = Callback()

    # Initialize the speech synthesizer
    # you can customize the synthesis parameters, like voice, format, sample_rate or other parameters
    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__':
    # 必須先設定 dashscope.api_key 和 base_websocket_api_url，再建立 SpeechSynthesizerObjectPool。
    # 池在初始化時即按當前全域 dashscope.api_key 建立 WebSocket 串連，
    # 池建立後再修改 dashscope.api_key 不會影響池內已建串連。
    # 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
    dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'
    init_dashscope_api_key()

    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))

    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()

資源管理與異常處理

• 任務成功：當語音合成任務正常完成時，必須調用 connectionPool.return_synthesizer(speech_synthesizer) 將 SpeechSynthesizer 對象歸還到池中，以便複用。

  重要不要歸還未完成任務或任務失敗的SpeechSynthesizer對象。

• 任務失敗：當 SDK 內部或商務邏輯拋出異常導致任務中斷時，主動關閉底層的 WebSocket 串連：speech_synthesizer.close()

• 在所有語音合成任務完成後，要通過如下方式關閉對象池：connectionPool.shutdown()。

• 在服務出現TaskFailed報錯時，不需要額外處理。

Java SDK

Java SDK通過內建的串連池和自訂的對象池協同工作，實現最佳效能。

• 串連池：SDK 內部整合的 OkHttp3 串連池，負責管理和複用底層的 WebSocket 串連，減少網路握手開銷。此功能預設開啟。
• 對象池：基於 commons-pool2 實現，用於維護一組已預先建立好串連的 SpeechSynthesizer 對象。從池中擷取對象可消除串連建立的延遲，顯著降低首包延遲。

實現步驟

1. 添加依賴

   根據專案構建工具，在依賴設定檔中添加 dashscope-sdk-java 和 commons-pool2。

   以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>
   

   1. 儲存pom.xml檔案。
   2. 使用Maven命令（如mvn clean install或mvn compile）來更新專案依賴

   Gradle

   1. 開啟Gradle專案的build.gradle檔案。
   2. 在dependencies塊內添加以下依賴資訊。

   dependencies {
       // 請將 'the-latest-version' 替換為2.16.6及以上版本，可在如下連結查詢相關版本號碼：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
   

2. 配置串連池

   通過環境變數配置串連池關鍵參數：

   環境變數	描述
   DASHSCOPE_CONNECTION_POOL_SIZE	串連池大小。 推薦值：峰值並發數的 2 倍以上。 預設值：32。
   DASHSCOPE_MAXIMUM_ASYNC_REQUESTS	最大非同步請求數。 推薦值：與 DASHSCOPE_CONNECTION_POOL_SIZE 保持一致。 預設值：32。
   DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST	單主機最大非同步請求數。 推薦值：與 DASHSCOPE_CONNECTION_POOL_SIZE 保持一致。 預設值：32。

3. 設定物件池

   通過環境變數設定物件池大小：

   環境變數	描述
   COSYVOICE_OBJECTPOOL_SIZE	對象池大小。 推薦值：峰值並發數的 1.5 至 2 倍。 預設值：500。

   重要

   • 對象池的大小（COSYVOICE_OBJECTPOOL_SIZE）必須小於或等於串連池的大小（DASHSCOPE_CONNECTION_POOL_SIZE）。否則，當對象池請求對象時，若串連池已滿，會導致調用線程阻塞，等待可用串連。
   • 對象池大小不應超過賬戶的 QPS（每秒查詢率）限制。

   通過如下代碼建立對象池：

class CosyvoiceObjectPool {
    // 。。。這裡省略其它代碼，完整樣本請參見完整代碼
    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;
    }
}

4. 從對象池中擷取SpeechSynthesizer對象

   如果當前未歸還的對象數量已超過對象池的最大容量，系統會額外建立一個新的SpeechSynthesizer對象。

   此類新建立的對象需要重新進行初始化並建立 WebSocket 串連，無法利用對象池的既有串連資源，因此不具備複用效果。

synthesizer = CosyvoiceObjectPool.getInstance().borrowObject();

5. 進行語音合成

   從對象池借出SpeechSynthesizer對象後，需先調用updateParamAndCallback(param, callback)關聯本次任務的參數與回調，再調用streamingCall或call方法進行語音合成。

   重要

   • 在對象池情境中，updateParamAndCallback會被多次調用（每次借出對象時都需調用一次，用於切換該次任務的回調和任務級參數，如voice、format等）。多次調用時傳入的apiKey必須始終相同。updateParamAndCallback只更新當前SpeechSynthesizer執行個體的本地欄位，不會重建底層 WebSocket 串連；而 SDK 僅在 WebSocket 建連握手時將apiKey寫入Authorization要求標頭用於鑒權，後續任務訊息（如run-task）本身不攜帶apiKey。因此只要複用的串連未斷開，傳入新的apiKey不會被發送到服務端，請求實際仍會使用串連首次握手時的apiKey，可能導致身份、配額或計費歸屬與預期不一致。
   • 如確需使用多個不同的 API Key，請為每個 API Key 維護獨立的對象池執行個體。

6. 歸還SpeechSynthesizer對象

   語音合成任務結束後，歸還SpeechSynthesizer對象，以便後續任務可以複用該對象。

   不要歸還未完成任務或任務失敗的對象。

CosyvoiceObjectPool.getInstance().returnObject(synthesizer);

完整代碼

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 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不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
                            // 若沒有配置環境變數，請用阿里雲百鍊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();
                // 注意：對象池情境下，多次調用 updateParamAndCallback 時傳入的 apiKey 必須始終相同；SDK 不會更新已建立串連的 apiKey，傳入不同的 apiKey 不會生效。詳見上文“進行語音合成”步驟的重要說明。
                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 {
        // 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
        Constants.baseWebsocketApiUrl = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference";
        // 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);
    }
}

推薦配置

以下配置基於在指定規格的阿里雲伺服器上僅運行 CosyVoice 語音合成服務的測試結果。過高的並發數可能導致任務處理延遲。

其中單機並發數指的是同一時刻正在啟動並執行CosyVoice語音合成任務數，也可以理解為背景工作執行緒數。

資源管理與異常處理

• 任務成功：當語音合成任務正常完成時，必須調用GenericObjectPool的returnObject方法將SpeechSynthesizer對象歸還到池中，以便複用。

  在當前代碼中，對應CosyvoiceObjectPool.getInstance().returnObject(synthesizer)

  重要不要歸還未完成任務或任務失敗的SpeechSynthesizer對象。

• 任務失敗：當 SDK 內部或商務邏輯拋出異常導致任務中斷時，必須執行以下兩個操作：

  1. 主動關閉底層的 WebSocket 串連
  2. 從對象池中廢棄該對象，防止被再次使用

// 在當前代碼中對應如下內容
// 關閉串連
synthesizer.getDuplexApi().close(1000, "bye");
// 在對象池中廢棄出現異常的synthesizer
CosyvoiceObjectPool.getInstance().invalidateObject(synthesizer);

• 在服務出現TaskFailed報錯時，不需要額外處理。

調用預熱與耗時統計說明

在對 DashScope Java SDK 進行並發調用延遲等效能評估時，建議先執行充分的預熱操作，確保測量結果反映穩定點下的真實效能，避免初始串連耗時導致資料偏差。

串連複用機制

DashScope Java SDK 通過全域單例的串連池高效管理和複用 WebSocket 串連，旨在減少頻繁建連和斷連的開銷，提升高並發情境下的處理能力。

該機制的工作特點如下：

• 按需建立：SDK 不會在服務啟動時預建立 WebSocket 串連，而是在首次調用時按需建立。

• 限時複用：請求完成後，串連將在池中保留最多 60 秒以備複用。

  • 若 60 秒內有新請求，將複用現有串連，避免重複握手開銷。
  • 若串連空閑超過 60 秒，將被自動關閉以釋放資源。

預熱的重要性

在以下情境中，串連池中可能沒有可複用的活躍串連，導致請求需要建立串連：

• 應用剛啟動，尚未發起任何調用。
• 服務空閑時間超過 60 秒，池中串連已因逾時而關閉。

在這些情境下，首次請求需完成 WebSocket 建連（TCP 握手、TLS 協商、協議升級），延遲顯著高於後續複用串連的請求。若未預熱，效能測試結果會因包含建連耗時而產生偏差。

SDK側延遲與實際首包延遲的區別

SDK側列印的首包延遲（如通過 get_first_package_delay() 擷取的值）包含了 WebSocket 建聯和網路傳輸等耗時，並不等同於模型服務的實際首包延遲。

實際首包延遲是指從服務端收到 run-task 指令到返回第一個 result-generated 事件的時間間隔，該值可通過服務端日誌查看。

在高並發情境下，由於大量串連的建立和資源調度，SDK側列印的延遲數值可能顯著高於服務端的實際首包延遲。如果觀察到 SDK 報告的首包延遲較高，建議：

• 對比服務端日誌中的首包延遲（從 run-task 到首個 result-generated），確認模型推理效能是否正常。
• 使用上述對象池或串連池機制進行預熱，消除 WebSocket 建連開銷，使 SDK 側列印的延遲更接近實際首包延遲。

推薦做法

為擷取可靠的效能資料，在正式進行效能壓測或延遲統計前，請遵循以下預熱步驟：

1. 類比正式測試的並發層級，提前發起一定數量的調用（例如，持續 1-2 分鐘），以充分填充串連池。
2. 確認串連池已建立並維持足夠的活躍串連後，再開始正式的效能資料採集。

通過合理的預熱，可使 SDK 串連池進入穩定複用狀態，從而測量出更具代表性的延遲指標，真實反映服務線上上平穩運行時的效能。

Java SDK常見異常