一种FreeSWITCH流量镜像WebSocket音频推流方案

赴忽 · 2025-8-17 03:27:30

环境：CentOS 7.9_x64
FreeSWITCH版本： 1.10.11（docker 23.0.6部署）
Python版本：3.9.12
之前遇到过一个需求：
在不影响生产系统业务（FreeSWITCH服务）的情况下，实时推送WebSocket格式语音流给解析服务器，以完成后续服务（比如实时辅助、实时质检等）。
今天整理下该需求的可行性验证笔记并提供示例代码，如需商业使用请参考文章自行实现。
本文使用的FreeSWITCH是基于docker部署，如有使用docker部署FreeSWITCH的需求，可参考如下文章：
https://www.cnblogs.com/MikeZhang/p/dockerPackFs20230617.html一、整体结构

为了演示方便，本文使用基于主机的流量镜像方案，生产环境可使用交换机的镜像口来替换。
此外，关于编码方面，这里做下简化：
1）流量镜像rtp编码是g711a（pcma）；
2）ws推流的目标编码是 L16（16k）；

整体结构如下：

二、流量镜像环境准备及策略实现

1、环境说明

整体结构如下：

说明：
1）192.168.137.100机器部署freeswitch，并配置流量镜像推流；
2）192.168.137.101机器部署流量镜像推流程序、ws模拟服务程序；
2、推流策略实现及效果验证

这里使用iptables的tee模块实现流量镜像功能，为了演示方便，这里使用udp流量全推的方案。
1）推流策略配置
流量镜像配置示例：

[root@host100 pbx]# cat /root/tee1.sh
#! /bin/bash
iptables -t mangle -F
iptables -t mangle -I PREROUTING 1 -p udp -j TEE --gateway 192.168.137.101
iptables -t mangle -I POSTROUTING 1 -p udp -j TEE --gateway 192.168.137.101
iptables -t mangle -L -v
[root@host100 pbx]#

复制代码

2）推流效果验证
在100机器上执行如下命令：

nc -u 192.168.137.108 33330

复制代码

输入 123456 ，然后执行 ctrl + d 键进行发送。
在101机器上进行抓包：

tcpdump -i enp0s3 udp port 33330 -vvvvvvvvvvvvvvvv

复制代码

使用tcpdump验证效果如下：

使用wireshark验证效果如下：

三、搭建ws验证服务器

为了方便方案验证，这里使用python模拟个ws服务器，该服务器会将收到的音频数据存储为本地pcm文件以便验证。
这里用的python3.9环境，环境部署可参考如下文章：
https://www.cnblogs.com/MikeZhang/p/centos7-install-py39-20220704.html配合验证的ws服务器代码如下（wsServer2.py）：

import os,json,time
from websockets.sync.server import serve
gPcmDir="pcmFiles"
def handler(ws):
callid,tag = "",0
fout,fName = None,""
clientId = id(ws)
for msg in ws:
#print("%s,len msg : %d "% (fName,len(msg)))
#msg2 = "I got your message: {}".format(msg)
#websocket.send(msg)
if type(msg) == str:
#if msg.find("initConfig") > -1:
print(msg)
dtmp = json.loads(msg)
var1 = dtmp.get("call_id")
if var1 : call_id = var1
var2 = dtmp.get("tag")
if None != var2 : tag = var2
print(call_id,tag)
#fName = "%s-%d.pcm" % (call_id,role)
fName = "%s/%s.pcm" % (gPcmDir,tag)
if None == fout : fout = open(fName,"wb")
if "end" == dtmp.get("type"):
fout.close()
if type(msg) == bytes:
fout.write(msg)
return None
def main():
if not os.path.isdir(gPcmDir):os.makedirs(gPcmDir)
with serve(handler, "0.0.0.0", 8765) as server:
server.serve_forever()
if __name__ == "__main__":
main()

复制代码

启动脚本如下（start.sh）

#! /bin/bash
#pydir=$PWD
pydir=/root/python39
export CFLAGS="-I$pydir/include"
export LDFLAGS="-L$pydir/lib"
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$pydir/lib
$pydir/bin/python3.9 wsServer2.py

复制代码

运行效果如下：

四、核心功能实现

这里列举下核心功能，并提供python示例代码。
1、从抓包数据中提取alaw音频

关键点如下：
1）抓包功能通过libpcap实现；
2）解析sdp获取rtp端口信息；
3）根据rtp协议从payload中解析并提取音频数据；
4）使用ffmpeg将提取的alaw格式音频转换为wav格式，以便验证音频的正确性；
示例代码如下（pcap2alaw.py）：

完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。2、alaw音频转换成pcm音频（L16）

alaw2pcm示例代码：

import os,struct,socket
import ctypes as ct
import audioop
rawFile = b"g711a_export1.raw"
sampwidth = 4
i,step = 0,160//sampwidth
fout = open("pcm_export1.raw","wb")
with open(rawFile,"rb") as fin:
while True:
data = fin.read(step)
if not data : break
data2 = audioop.alaw2lin(data, sampwidth)
fout.write(data2)

复制代码

提取的音频导入效果如下：

关于pcm音频的播放可参考如下文章：
https://www.cnblogs.com/MikeZhang/p/pcm20232330.html3、pcm音频以ws格式发送

pcm转ws示例代码如下（pcm2ws.py）：

完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。运行效果如下：

五、WebSocket音频推流方案

这里描述下独立程序的推流方案，即：所有功能都在一个程序里面实现，比如单独的python脚本。
1、模块设计

模块设计如下：

2、程序设计

大致设计思路如下：
1）启动独立线程ThrdLiveCap，使用libpcap实时抓取流量镜像过来的数据，并将数据推送到指定队列；
2）启动独立线程ThrdPktParse，解析sip协议，提取uuid、rtp端口等关键信息，并将解析的数据存储到全局缓存中；
3）启动独立线程ThrdRtp2Ws，监测并启动独立线程，进行ws推流；
3、示例代码

这里提供下示例代码。

完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。4、运行效果

1）编写拨号方案
内容如下：

<extension name="playTest">
<condition field="destination_number" expression="^10087$">
</condition>
</extension>

复制代码

2）注册1000分机，拨打10087号码，在101机器上启动rtp2ws脚本、wsServer脚本，可成功接收rtp文件。
运行效果如下：

文件导入效果如下（s16le 16K）：

可正常播放。
对应的pcm文件可从文末提供的渠道获取。
六、资源下载

本文相关资源及运行环境，可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

一种FreeSWITCH流量镜像WebSocket音频推流方案

浏览过的版块

签约作者