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 キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
	// 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: 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("Connection failed, HTTP status code: %d\n", resp.StatusCode)
		}
		fmt.Println("Connection failed:", err)
		return
	}
	defer conn.Close()

	// タスク ID を生成します
	taskID := uuid.New().String()
	fmt.Printf("Generated task 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 イベントを 1 回しか送信できません。それ以外の場合、「Text request limit violated, expected 1.」というエラーが返されます
				"enable_ssml": false,
			},
			"input": map[string]interface{}{},
		},
	}

	runTaskJSON, _ := json.Marshal(runTaskCmd)
	fmt.Printf("Sending run-task event: %s\n", string(runTaskJSON))

	err = conn.WriteMessage(websocket.TextMessage, runTaskJSON)
	if err != nil {
		fmt.Println("Failed to send run-task:", err)
		return
	}

	textSent := false

	// メッセージを処理します
	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			fmt.Println("Failed to read message:", err)
			break
		}

		// バイナリメッセージを処理します
		if messageType == websocket.BinaryMessage {
			fmt.Printf("Received binary message, length: %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("Received text message: %s\n", strings.ReplaceAll(messageStr, "\n", ""))

		// JSON を解析してイベントタイプを取得します
		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("Event type: %s\n", event)

					switch event {
					case "task-started":
						fmt.Println("=== Received task-started event ===")

						if !textSent {
							// continue-task イベントを送信します

							texts := []string{"Before my bed, moonlight shines bright, I suspect it's frost upon the ground.", "I raise my eyes to gaze at the bright moon, then bow my head, thinking of home."}

							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("Sending continue-task event: %s\n", string(continueTaskJSON))

								err = conn.WriteMessage(websocket.TextMessage, continueTaskJSON)
								if err != nil {
									fmt.Println("Failed to send 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("Sending finish-task event: %s\n", string(finishTaskJSON))

							err = conn.WriteMessage(websocket.TextMessage, finishTaskJSON)
							if err != nil {
								fmt.Println("Failed to send finish-task:", err)
								return
							}
						}

					case "task-finished":
						fmt.Println("=== Task completed ===")
						return

					case "task-failed":
						fmt.Println("=== Task failed ===")
						if header["error_message"] != nil {
							fmt.Printf("Error message: %s\n", header["error_message"])
						}
						return

					case "result-generated":
						fmt.Println("Received result-generated event")
					}
				}
			}
		}
	}
}

C#

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

class Program {
    // API キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    // 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: 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 = {
                "Before my bed, moonlight shines bright,",
                "I suspect it\'s frost upon the ground.",
                "I raise my eyes to gaze at the bright moon,",
                "then bow my head, thinking of home."
            };
            foreach (string text in texts) {
                await SendContinueTaskCommandAsync(text);
            }

            // finish-task イベントを送信します
            await SendFinishTaskCommandAsync(_taskId);

            // 受信タスクの完了を待ちます
            await receiveTask;

            Console.WriteLine("Task completed, connection closed.");
        } catch (OperationCanceledException) {
            Console.WriteLine("Task cancelled.");
        } catch (Exception ex) {
            Console.WriteLine($"Error occurred: {ex.Message}");
        } finally {
            _cancellationTokenSource.Cancel();
            _webSocket.Dispose();
        }
    }

    private static void ClearOutputFile(string filePath) {
        if (File.Exists(filePath)) {
            File.WriteAllText(filePath, string.Empty);
            Console.WriteLine("Output file cleared.");
        } else {
            Console.WriteLine("Output file does not exist, no need to clear.");
        }
    }

    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("Successfully connected to the WebSocket service.");
        } catch (OperationCanceledException) {
            Console.WriteLine("WebSocket connection cancelled.");
        } catch (Exception ex) {
            Console.WriteLine($"WebSocket connection failed: {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 イベントを 1 回しか送信できません。それ以外の場合、「Text request limit violated, expected 1.」というエラーが返されます
                enable_ssml = false
            },
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("run-task event sent.");
    }

    private static async Task SendContinueTaskCommandAsync(string text) {
        if (_taskId == null) {
            throw new InvalidOperationException("Task ID not initialized.");
        }

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

        await SendJsonMessageAsync(command);
        Console.WriteLine("continue-task event sent.");
    }

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

        await SendJsonMessageAsync(command);
        Console.WriteLine("finish-task event sent.");
    }

    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("Message sending cancelled.");
        }
    }

    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("Task started.");
                        _taskStartedTcs.TrySetResult(true);
                        break;
                    case "task-finished":
                        Console.WriteLine("Task completed.");
                        _cancellationTokenSource.Cancel();
                        break;
                    case "task-failed":
                        Console.WriteLine("Task failed: " + 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("Received binary data...");

                // バイナリデータをファイルに保存します
                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("Message receiving cancelled.");
            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 キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
// 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: $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 サーバー URL
$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' => 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 "Connected to the WebSocket server\n";

    // タスク ID を生成します
    $taskId = generateTaskId();

    // run-task イベントを送信します
    sendRunTaskMessage($conn, $taskId);

    // continue-task イベントを送信する関数を定義します
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
        // 送信するテキスト
        $texts = ["Before my bed, moonlight shines bright,", "I suspect it\'s frost upon the ground.", "I raise my eyes to gaze at the bright moon,", "then bow my head, thinking of home."];
        $continueTaskCount = 0;
        foreach ($texts as $text) {
            $continueTaskMessage = json_encode([
                "header" => [
                    "action" => "continue-task",
                    "task_id" => $taskId,
                    "streaming" => "duplex"
                ],
                "payload" => [
                    "input" => [
                        "text" => $text
                    ]
                ]
            ]);
            echo "Sending continue-task event: " . $continueTaskMessage . "\n";
            $conn->send($continueTaskMessage);
            $continueTaskCount++;
        }
        echo "Number of continue-task events sent: " . $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 "Unknown message format\n";
            }
        }
    });

    // 接続クローズをリッスンします
    $conn->on('close', function($code = null, $reason = null) {
        echo "Connection closed\n";
        if ($code !== null) {
            echo "Close code: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "Close reason: " . $reason . "\n";
        }
    });
}, function ($e) {
    echo "Unable to connect: {$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 イベントを 1 回しか送信できません。それ以外の場合、「Text request limit violated, expected 1.」というエラーが返されます
                "enable_ssml" => false
            ],
            "input" => (object) []
        ]
    ]);
    echo "Sending run-task event: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task event sent\n";
}

/**
 * オーディオファイルを読み込みます
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "Unable to read audio file\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 "Sending finish-task event: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task event sent\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 "Task started, sending continue-task events...\n";
            $taskStarted = true;
            // continue-task イベントを送信します
            $sendContinueTask();
            break;
        case 'result-generated':
            // result-generated イベントを受信しました
            break;
        case 'task-finished':
            echo "Task completed\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "Task failed\n";
            echo "Error code: " . $response['header']['error_code'] . "\n";
            echo "Error message: " . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "Error: " . $response['payload']['message'] . "\n";
            break;
        default:
            echo "Unknown event: " . $response['header']['event'] . "\n";
            break;
    }

    // タスクが完了した場合、接続をクローズします
    if ($response['header']['event'] == 'task-finished') {
        // すべてのデータが送信されたことを確認するために 1 秒待ちます
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "Client closing connection\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 キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
// 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: 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('Connected to the WebSocket server');

  // 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 イベントを 1 回しか送信できません。それ以外の場合、「Text request limit violated, expected 1.」というエラーが返されます
      },
      input: {}
    }
  });
  ws.send(runTaskMessage);
  console.log('run-task message sent');
});

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('Task started');
        // continue-task イベントを送信します
        sendContinueTasks(ws);
        break;
      case 'task-finished':
        console.log('Task completed');
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      case 'task-failed':
        console.error('Task failed: ', message.header.error_message);
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      default:
        // result-generated はここで処理できます
        break;
    }
  }
});

function sendContinueTasks(ws) {
  const texts = [
    'Before my bed, moonlight shines bright,',
    'I suspect it is frost upon the ground.',
    'I raise my eyes to gaze at the bright moon,',
    'then bow my head, thinking of home.'
  ];

  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 sent, text: ${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 sent');
    }
  }, texts.length * 1000 + 1000); // すべての continue-task イベント送信後 1 秒で送信します
}

ws.on('close', () => {
  console.log('Disconnected from the WebSocket server');
});

Java

開発には Java DashScope SDK の使用を推奨します。「Qwen-Audio-TTS Java SDK / CosyVoice 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("Connection established");

        // run-task イベントを送信します
        // enable_ssml を true に設定すると、continue-task イベントを 1 回しか送信できません。それ以外の場合、「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("Received message from server: " + message);
        try {
            // JSON メッセージを解析します
            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("Received task-started event from server");

                        List<String> texts = Arrays.asList(
                                "Before my bed, moonlight shines bright,I suspect it\'s frost upon the ground.",
                                "I raise my eyes to gaze at the bright moon,then bow my head, thinking of home."
                        );

                        for (String text : texts) {
                            // continue-task イベントを送信します
                            sendContinueTask(text);
                        }

                        // finish-task イベントを送信します
                        sendFinishTask();
                    } else if ("task-finished".equals(event)) {
                        System.out.println("Received task-finished event from server");
                        taskFinished = true;
                        closeConnection();
                    } else if ("task-failed".equals(event)) {
                        System.out.println("Task failed: " + message);
                        closeConnection();
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("Exception occurred: " + e.getMessage());
        }
    }

    @Override
    public void onMessage(ByteBuffer message) {
        System.out.println("Received binary audio data of size: " + message.remaining());

        try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
            byte[] buffer = new byte[message.remaining()];
            message.get(buffer);
            fos.write(buffer);
            System.out.println("Audio data written to local file " + outputFile);
        } catch (IOException e) {
            System.err.println("Failed to write audio data to local file: " + e.getMessage());
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("Connection closed: " + reason + " (" + code + ")");
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("Error: " + 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 キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
            // 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: String apiKey = "sk-xxx"
            String apiKey = System.getenv("DASHSCOPE_API_KEY");
            if (apiKey == null || apiKey.isEmpty()) {
                System.err.println("Please set the DASHSCOPE_API_KEY environment variable");
                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("Failed to connect to WebSocket service: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Python

開発には Python DashScope SDK の使用を推奨します。「Qwen-Audio-TTS Python SDK / CosyVoice 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 キー
        uri (str): WebSocket サービス URL
    """
        self.api_key = api_key  # ご自身の API キーに置き換えてください
        self.uri = uri  # ご自身の WebSocket URL に置き換えてください
        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 connected")

        # 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 イベントは 1 つしか送信できません。それ以外の場合はエラーが返されます。
                    "enable_ssml": False
                },
                "input": {}
            }
        }

        # run-task イベントを送信します
        ws.send(json.dumps(run_task_cmd))
        print("Sent run-task event")

    def on_message(self, ws, message):
        """
    メッセージを受信したときのコールバックです。
    テキストメッセージとバイナリメッセージを別々に処理します。
    """
        if isinstance(message, str):
            # JSON テキストメッセージを処理します
            try:
                msg_json = json.loads(message)
                print(f"Received JSON message: {msg_json}")

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

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

                        if event == "task-started":
                            print("Task started")
                            self.task_started = True

                            # continue-task イベントを送信します
                            texts = [
                                "Before my bed, moonlight shines bright,I suspect it\'s frost upon the ground.",
                                "I raise my eyes to gaze at the bright moon,then bow my head, thinking of home."
                            ]

                            for text in texts:
                                self.send_continue_task(text)

                            # すべての continue-task イベントを送信した後に finish-task を送信します
                            self.send_finish_task()

                        elif event == "task-finished":
                            print("Task completed")
                            self.task_finished = True
                            self.close(ws)

                        elif event == "task-failed":
                            error_msg = msg_json.get("error_message", "Unknown error")
                            print(f"Task failed: {error_msg}")
                            self.task_finished = True
                            self.close(ws)

            except json.JSONDecodeError as e:
                print(f"JSON parsing failed: {e}")
        else:
            # バイナリメッセージ (オーディオデータ) を処理します
            print(f"Received binary message, size: {len(message)} bytes")
            with open(self.output_file, "ab") as f:
                f.write(message)
            print(f"Audio data written to local file {self.output_file}")

    def on_error(self, ws, error):
        """エラーが発生したときのコールバックです。"""
        print(f"WebSocket error: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        """接続が閉じられたときのコールバックです。"""
        print(f"WebSocket closed: {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"Sent continue-task event, text content: {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("Sent finish-task event")

    def close(self, ws):
        """接続をアクティブに閉じます。"""
        if ws and ws.sock and ws.sock.connected:
            ws.close()
            print("Connection closed actively")

    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("Listening for WebSocket messages...")
        self.ws.run_forever()  # 持続的接続リスナーを開始します

# 使用例
if __name__ == "__main__":
    # API キーはシンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    # 環境変数を設定していない場合は、次の行を中国語版 Model Studio の API キーに置き換えてください: 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 URL に置き換えてください

    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

synthesizer_object_pool = SpeechSynthesizerObjectPool(max_size=20)
import dashscope
# 以下は中国 (北京) リージョンの構成です。{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 = [
    'Sentence 1: Welcome to Alibaba speech synthesis service.',
    'Sentence 2: Welcome to Alibaba speech synthesis service.',
    'Sentence 3: Welcome to Alibaba speech synthesis service.',
]
connectionPool = None

def init_dashscope_api_key():
    '''
    DashScope API キーを設定します。詳細情報:
    https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
    '''
    # API キーはシンガポールリージョンと北京リージョンで異なります。API キーの取得: 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']  # 環境変数 DASHSCOPE_API_KEY から API キーを読み込みます
    else:
        dashscope.api_key = '<your-dashscope-api-key>'  # API キーを手動で設定します

def synthesis_text_to_speech_and_play_by_streaming_mode(text, task_id):
    global USE_CONNECTION_POOL, connectionPool
    '''
    ストリーミングモードで指定されたテキストから音声を合成し、非同期呼び出しで合成された音声をリアルタイムで再生します。
    詳細については、https://www.alibabacloud.com/help/document_detail/2712523.html をご参照ください。
    '''

    complete_event = threading.Event()

    # 結果を処理するためのコールバックを定義します

    class Callback(ResultCallback):
        def on_open(self):
            # オブジェクトプールを使用する場合、on_open はタスク開始後に呼び出されます
            self.file = open(f'result_{task_id}.mp3', 'wb')
            print(f'[task_{task_id}] start')

        def on_complete(self):
            print(f'[task_{task_id}] 音声合成タスクが正常に完了しました。')
            complete_event.set()

        def on_error(self, message: str):
            print(f'[task_{task_id}] 音声合成タスクが失敗しました。{message}')

        def on_close(self):
            # オブジェクトプールを使用する場合、on_open はタスク終了後に呼び出されます
            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:
            # プレーヤーに送信します
            # 音声をファイルに保存します
            self.file.write(data)

    # 音声合成コールバックを呼び出します
    synthesizer_callback = Callback()

    # 音声合成器を初期化します
    # 音声、フォーマット、サンプルレートなどの合成パラメーターをカスタマイズできます
    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}] 音声合成タスクが失敗しました。{e}')
        if USE_CONNECTION_POOL:
            # 接続プールを使用する場合、タスクが失敗した場合は合成器の接続を手動で閉じます。
            speech_synthesizer.close()
        return

    print('[task_{}] 合成テキスト: {}'.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__':
    # SpeechSynthesizerObjectPool を作成する前に、dashscope.api_key と base_websocket_api_url を設定する必要があります。
    # プールは初期化時に現在のグローバル 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. mvn clean install または mvn compile などの Maven コマンドを実行して、プロジェクトの依存関係を更新します。

   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. 接続プールの構成

   環境変数を通じて主要な接続プールパラメーターを構成します:

3. オブジェクトプールの構成

   環境変数を通じてオブジェクトプールサイズを構成します:

以下のコードでオブジェクトプールを作成します:

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 を使用します。これにより、ID、クォータ、課金の帰属が期待とは異なる可能性があります。
   • 複数の API キーを使用する場合は、各キーに対して別々のオブジェクトプールインスタンスを維持してください。

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 com.alibaba.dashscope.utils.Constants;
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();
        // 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("Received audio, audio stream 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 キーはシンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
                            // 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: .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 " + Thread.currentThread() + "] Speech synthesis task completed. Time cost: " + 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";
        // 接続プール環境をチェックします
        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;
        // SpeechSynthesis オブジェクトのプールを作成します
        ExecutorService executorService = Executors.newFixedThreadPool(runTimes);

        for (int i = 0; i < runTimes; i++) {
            // タスク送信時刻を記録します
            LocalDateTime submissionTime = LocalDateTime.now();
            executorService.submit(new SynthesizeTaskWithCallback(new String[] {
                    "床前明月光，", "疑是地上霜，", "举头望明月，", "低头思故乡。"}));
        }

        // ExecutorService をシャットダウンし、すべてのタスクの完了を待ちます
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.MINUTES);
        System.exit(0);
    }
}

推奨構成

以下の構成は、指定された仕様の Alibaba Cloud ECS インスタンス上で Qwen-Audio-TTS/CosyVoice 音声合成サービスのみを実行したテスト結果に基づいています。過剰な同時実行はタスク処理の遅延を増加させる可能性があります。

「単一マシンの同時実行数」とは、Qwen-Audio-TTS/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 報告遅延 vs 実際の最初のパケット遅延

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 例外