VoiceSideStT RocketMQ 消息对接接口文档

{
  "uuid": "CALL-20260720-0001",
  "call_hash": "13800138000",
  "leg": "aleg",
  "offer": "stt_quality_topic",
  "content": "您好，请问需要什么帮助？",
  "date": "2026-07-20 10:23:45",
  "timestamp": "1779003825000",
  "emotion": "NEUTRAL",
  "speechRate": 3.8,
  "markedText": "<NEUTRAL>您好，请问需要什么帮助？",
  "voiceFile": "/record/20260720/CALL-20260720-0001/aleg/1779003825000.wav"
}

{
  "uuid": "CALL-20260720-0001",
  "call_hash": "13800138000",
  "leg": "bleg",
  "offer": "stt_quality_topic",
  "content": "我想咨询一下套餐资费",
  "date": "2026-07-20 10:23:48",
  "timestamp": "1779003828000"
}

{
  "uuid": "CALL-20260720-0002",
  "call_hash": "13900139000",
  "leg": "aleg",
  "offer": "stt_quality_topic",
  "content": "好的，马上为您办理。",
  "date": "2026-07-20 10:25:10",
  "timestamp": "1779003910000",
  "agentId": "8001",
  "deptCode": "sales"
}

import com.alibaba.fastjson.JSON;          // 或 com.google.gson.Gson
import com.google.gson.Gson;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class SideSttConsumer {

    public static void main(String[] args) throws Exception {
        String namesrvAddr = "127.0.0.1:9876";          // 由运维提供
        String topic       = "stt_quality_topic";        // 由运维提供
        String tag         = "sidestt";                  // 由运维提供
        String consumerGroup = "cid_sidestt_consumer";

        // 1. 构建 Consumer（若服务端未开启 ACL，去掉 AclClientRPCHook 参数即可）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                consumerGroup,
                new AclClientRPCHook(new SessionCredentials(
                        "YOUR_ACCESS_KEY",   // 由运维提供
                        "YOUR_SECRET_KEY"))); // 由运维提供

        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe(topic, tag);   // 订阅指定 Tag；接收全部 Tag 用 "*"

        // 2. JSON 解析器（生产端用 Gson 序列化，消费端用 Gson/fastjson 均可）
        Gson gson = new Gson();

        // 3. 注册监听
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                // ★ Body 为 UTF-8 字节流，务必按 UTF-8 解码
                String body = new String(msg.getBody(), "UTF-8");

                // 反序列化到与生产端对称的 POJO（字段名一致即可）
                SideSttMessage dto = gson.fromJson(body, SideSttMessage.class);

                // 业务处理：落库 / 智能填单 / 实时大屏 ...
                handle(dto);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        System.out.println("SideSTT Consumer started.");
    }

    private static void handle(SideSttMessage dto) {
        // TODO: 落库/转发/填单等业务逻辑
    }

    /** 与生产端 SideSttMqMessage（com.pbxbusiness.sidestt.json）字段对称的 POJO。 */
    static class SideSttMessage {
        private String uuid;
        private String call_hash;     // Gson 默认按字段名映射；若改名 callHash 需加 @SerializedName("call_hash")
        private String leg;
        private String offer;
        private String content;
        private String date;
        private String timestamp;
        private String emotion;       // 可空：缺失即无情绪数据
        private Float speechRate;     // 可空：缺失即无语速数据
        private String markedText;    // 可空：缺失即无带标记文本
        private String voiceFile;     // 可空：缺失即无当次录音路径

        // getter / setter 省略...
    }
}

[MQ] makeMessage( gbk to utf-8 ):
{"uuid":"...","call_hash":"...","leg":"aleg",...}