基于Java开发客户端音频采集播放UDP协议转发程序

Posted HiveDark

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了基于Java开发客户端音频采集播放UDP协议转发程序相关的知识，希望对你有一定的参考价值。

一、综述

学习使用Java开发语言做计算机音频数据采集、压缩、转发功能，从而实现双向通话功能。采集数据频率为8KHz、16bit、单通道、小端格式，数据转发采用G711A压缩传输。

二、音频采样率

1. 参考百度百科

为了测试语音通话，音频采样率为8KHz即可满足要求。

2. 在数字音频领域，常用的采样率有：

  • 8,000 Hz - 电话所用采样率, 对于人的说话已经足够
  • 11,025 Hz-AM调幅广播所用采样率
  • 22,050 Hz和24,000 Hz- FM调频广播所用采样率
  • 32,000 Hz - miniDV 数码视频 camcorder、DAT (LP mode)所用采样率
  • 44,100 Hz - 音频 CD, 也常用于 MPEG-1 音频（VCD, SVCD, MP3）所用采样率
  • 47,250 Hz - 商用 PCM 录音机所用采样率
  • 48,000 Hz - miniDV、数字电视、DVD、DAT、电影和专业音频所用的数字声音所用采样率
  • 50,000 Hz - 商用数字录音机所用采样率
  • 96,000 或者 192,000 Hz - DVD-Audio、一些 LPCM DVD 音轨、BD-ROM（蓝光盘）音轨、和 HD-DVD （高清晰度 DVD）音轨所用所用采样率
  • 2.8224 MHz - Direct Stream Digital 的 1 位 sigma-delta modulation 过程所用采样率。

三、开发环境

  • JDK: 1.8
  • MINA: 2.1.3
  • SpringBoot: 1.3.1.RELEASE

四、核心代码

为了记录学习流程，接下来仅介绍核心代码部分，具体完整程序参考源码
本项目主要的业务逻辑分为两个部分：一是音频采集部分，二是数据通讯转发和接收部分。为了保证两部分的业务松耦合，需要对通讯和采集播放做程序设计上分离。

1. Socket通讯组件开发

为了保证底层通讯的和上层业务的松耦合，通讯部分做以下程序设计：

  • （1）将通讯框架的数据统一转为ByteBuffer做进一步的业务处理。例如将Mina的IoBuffer数据转为ByteBuffer交付于上层业务处理，从而保证对Mina通讯框架的松耦合，为以后切换其他的通讯框架做准备（比如Netty）
  • （2）统一定义Socket connector的接口标准，方便扩展。
  • （3）为了减少音频采集部分对通讯Connector的过渡依赖，采用单例工厂模式保证对外提供的通讯服务，管理所有启动的socket服务及释放资源等

1.1 SocketMessageHandler

定义消息上层业务转发统一接口，需要音频采集处理部分传递该处理部分。

package com.david.test.socket;

import java.nio.ByteBuffer;

public interface SocketMessageHandler 
	
	/**
	 * 获取消息
	 * @param byteBuffer
	 */
	public void onMessage(ByteBuffer byteBuffer);

1.2 SocketConnector

该类定义一个底层通讯的Connector的一些对外统一接口。

package com.david.test.socket;

import java.nio.ByteBuffer;

public interface SocketConnector 
	/**
	 * 获取Connector ID
	 * @return
	 */
	public String getId();
	/**
	 * 启动一个客户端
	 * @throws Exception
	 */
	public void startClient(SocketConnectorParams socketConnectorParams,SocketMessageHandler socketMessageHandler) throws Exception;
	
	/**
	 * 关闭客户端
	 * @throws Exception
	 */
	public void stopClient() throws Exception;
	
	/**
	 * 发送消息
	 * @param byteBuffer
	 * @throws Exception
	 */
	public void sendMessage(ByteBuffer byteBuffer) throws Exception;
	
	/**
	 * 连接是否存在
	 * @throws Exception
	 */
	public boolean isActive() throws Exception;

1.3 DefaultMinaUDPSocketConnector

基于Mina实现的缺省的UDP通讯Connector，源码如下：

package com.david.test.socket.connector;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.Date;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoConnector;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.iosession;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.proxy.utils.ByteUtilities;
import org.apache.mina.transport.socket.nio.NioDatagramConnector;
import org.springframework.util.StringUtils;

import com.david.test.socket.SocketConnector;
import com.david.test.socket.SocketConnectorParams;
import com.david.test.socket.SocketMessageHandler;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class DefaultMinaUDPSocketConnector implements SocketConnector
	private IoConnector connector;
	private IoSession ioSession;
	private String connectorId;
	
	@Override
	public String getId() 
		// TODO Auto-generated method stub
		return this.connectorId;
	

	@Override
	public boolean isActive() throws Exception 
		// TODO Auto-generated method stub
		return null == ioSession?false:ioSession.isActive();
	

	@Override
	public void stopClient() throws Exception 
		// TODO Auto-generated method stub
		try 
			if(null != ioSession)
				ioSession.closeNow();
				ioSession = null;
			
			if(null != connector)
				connector.dispose();
				connector = null;
			
			log.info("STOP UDP CLIENT SUCCESS, ID:",this.connectorId);
		 catch (Exception e) 
			log.error("Stop UDP Client Error", e);
		
	

	@Override
	public void startClient(SocketConnectorParams socketConnectorParams,SocketMessageHandler socketMessageHandler) throws Exception 
		// TODO Auto-generated method stub
		if(null == socketConnectorParams || StringUtils.isEmpty(socketConnectorParams.getHost()) 
				|| socketConnectorParams.getPort() < 1)
			throw new Exception("参数配置有误");
		
		if(null == socketMessageHandler)
			throw new Exception("Socket Message Handler 消息解析器配置异常");
		
		//connector Id
		this.connectorId = Thread.currentThread().getName()+"_"+new Date().getTime();
		try 
			InetSocketAddress inetSocketAddress = new InetSocketAddress(socketConnectorParams.getHost(), socketConnectorParams.getPort());
			connector = new NioDatagramConnector();
			connector.getFilterChain().addLast("logger", new LoggingFilter());
			connector.setHandler(new MinaUDPIoHandler(socketConnectorParams,socketMessageHandler));
			ConnectFuture connectFuture = connector.connect(inetSocketAddress);
			// 等待是否连接成功，相当于是转异步执行为同步执行。
			connectFuture.awaitUninterruptibly();
			log.info("Mina start UDP Client SUCCESS, ID:",this.connectorId);
		 catch (Exception e) 
			log.error("Mina start UDP Client Error", e);
		
	
	
	@Override
	public void sendMessage(ByteBuffer byteBuffer) throws Exception 
		// TODO Auto-generated method stub
		if(null == ioSession)
			throw new Exception("UDP session is released");
		
		ioSession.write(IoBuffer.wrap(byteBuffer.array()));
		log.info("ID:,SOCKET SEND:",this.connectorId,ByteUtilities.asHex(byteBuffer.array()).toUpperCase());
	
	
	// Mina UDP 数据处理
	class MinaUDPIoHandler extends IoHandlerAdapter
		private SocketConnectorParams socketConnectorParams;
		private SocketMessageHandler socketMessageHandler;
		
		public MinaUDPIoHandler(SocketConnectorParams socketConnectorParams, SocketMessageHandler socketMessageHandler) 
			this.socketConnectorParams = socketConnectorParams;
			this.socketMessageHandler = socketMessageHandler;
		

		@Override
		public void sessionOpened(IoSession session) throws Exception 
			// TODO Auto-generated method stub
			super.sessionOpened(session);
			ioSession = session;
		

		@Override
		public void sessionIdle(IoSession session, IdleStatus status) throws Exception 
			// TODO Auto-generated method stub
			super.sessionIdle(session, status);
		

		@Override
		public void exceptionCaught(IoSession session, Throwable cause) throws Exception 
			// TODO Auto-generated method stub
			super.exceptionCaught(session, cause);
		

		@Override
		public void messageReceived(IoSession session, Object message) throws Exception 
			// TODO Auto-generated method stub
			super.messageReceived(session, message);
			try 
				IoBuffer ioBuffer = (IoBuffer)message;
				int capacity = ioBuffer.capacity();
				int limit = ioBuffer.limit();
				byte[] data = new byte[limit];
				ioBuffer.get(data, 0, limit);
				log.info("RECV DATA: capacity[] limit[] data:",capacity,limit,ByteUtilities.asHex(data).toUpperCase());
				if(null != this.socketMessageHandler)
					this.socketMessageHandler.onMessage(ByteBuffer.wrap(data));
				
			 catch (Exception e) 
				log.error("Mina UDP RECV Message Error", e);
			
		

		@Override
		public void messageSent(IoSession session, Object message) throws Exception 
			// TODO Auto-generated method stub
			super.messageSent(session, message);

1.4 SocketConnectorFactory

统一管理系统中的所有的底层通讯Connector，提供服务。

package com.david.test.socket;
import java.util.LinkedList;
import java.util.List;

import com.david.test.socket.connector.DefaultMinaUDPSocketConnector;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class SocketConnectorFactory 
	private static SocketConnectorFactory socketConnectorFactory = new SocketConnectorFactory();
	private static List<SocketConnector> socketConnectors = new LinkedList<>();
	
	public static SocketConnectorFactory getInstance()
		return socketConnectorFactory;
	
	
	/**
	 * 创建一个UDPClient
	 * @return
	 */
	public SocketConnector createUDPClient()
		SocketConnector socketConnector = new DefaultMinaUDPSocketConnector();
		socketConnectors.add(socketConnector);
		return socketConnector;
	
	
	/**
	 * 启动一个UDPClient
	 * @param socketConnectorParams
	 * @param socketMessageHandler
	 * @return
	 * @throws Exception
	 */
	public SocketConnector startUDPClient(SocketConnectorParams socketConnectorParams,
			SocketMessageHandler socketMessageHandler) throws Exception
		SocketConnector socketConnector = new DefaultMinaUDPSocketConnector();
		socketConnector.startClient(socketConnectorParams, socketMessageHandler);
		socketConnectors.add(socketConnector);
		return socketConnector;
	
	
	/**
	 * 根据Connector ID移除Socket连接
	 * @param id
	 * @throws Exception
	 */
	public void releaseSocketConnector(String id) throws Exception
		for(SocketConnector socketConnector:socketConnectors)
			try 
				if(socketConnector.getId().equals(id))
					socketConnector.stopClient();
					socketConnectors.remove(socketConnector);
					break;
				
			 catch (Exception e) 
				log.error("关闭客户端异常", e);
			
		
		log.info("移除Socket监听器,剩余socket监听器数量:",socketConnectors.size());
	
	
	/**
	 * 释放所有的socket连接
	 * @throws Exception
	 */
	public void releaseAllSocketConnectors() throws Exception
		List<SocketConnector> toRemovedSocketConnectors = new LinkedList<>();
		for(SocketConnector socketConnector:socketConnectors)
			try 
				socketConnector.stopClient();
				toRemovedSocketConnectors.add(socketConnector);
			 catch (Exception e) 
				log.error("关闭客户端异常", e);
			
		
		socketConnectors.removeAll(toRemovedSocketConnectors);
		log.info("移除Socket监听器数量:,剩余socket监听器数量:",
				toRemovedSocketConnectors.size(),socketConnectors.size());

1.5 小结

以上1.1-1.4部分即完成了通讯服务的业务处理规则，此时还不涉及到业务开发，仅为通讯部分数据传输的统一标准。方便以后的底层通讯框架的切换等。

2. 音频数据采集转发和接收播放

音频采集是一个持续的过程，需要在子线程中处理。整体的思路如下：采集转发在子线程中处理，接收播放在主线程中处理即可（因为接收播放是异步通讯框架传递过来的数据）

2.1 UserAudioMedia

定义音频采集部分的规范，因为音频的采样率不同及压缩方式不同会存在不同的实现方式，定义统一的接口，便于扩展。

  • SocketMessageHandler: 该接口为和通讯框架之间的数据传递规范
  • Runnable: 该接口实现线程的Runnable接口，用于采集计算机音频数据转发
package com.david.test.audio;

import com.david.test.socket.SocketMessageHandler;

public interface UserAudioMedia extends SocketMessageHandler,Runnable
	/**
	 * 开始
	 */
	public void start();
	
	/**
	 * 停止
	 */
	public void stop();

2.2 DefaultUserAudioMedia

PCM原始音频流数据采集转发及接收播放

package com.david.test.audio.media;

import java.io.IOException;
import java.nio.ByteBuffer;

import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioInputStream;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.FloatControl;
import javax.sound.sampled.SourceDataLine;
import javax.sound.sampled.TargetDataLine;

import com.david.test.audio.UserAudioMedia;
import com.david.test.socket.SocketConnector;
import com.david.test.socket.SocketConnectorFactory;
import com.david.test.socket.SocketConnectorParams;
import com.david.test.utils.GsonUtils;

import lombok.extern.slf4j.Slf4j;
/**
 * 传输：PCM传输
 * 播放：PCM原始流播放
* @author 作者 :david E-mail:857332533@qq.com 
* @date 创建时间：2020年4月18日 上午10:24:37 
* @version 1.0 
* @since 2020年4月18日 上午10:24:37
 */
@Slf4j
public class DefaultUserAudioMedia implements UserAudioMedia
	private SocketConnectorParams socketConnectorParams;
	private RegistParam registParam;
	private SocketConnector socketConnector;
	private boolean isStop;
	private SourceDataLine sourceDataLine;
	private TargetDataLine targetDataLine;
	private Thread curThread = null;

	public DefaultUserAudioMedia(SocketConnectorParams socketConnectorParams,RegistParam registParam) throws Exception
		this.socketConnectorParams = socketConnectorParams;
		this.registParam = registParam;
		this.socketConnector = SocketConnectorFactory.getInstance().startUDPClient(socketConnectorParams, this);
		
	
	
	/**
	 * 停止
	 */
	@Override
	public void stop()
		this.isStop = true;
	
	
	/**
	 * 发起注册监听
	 * @param registParam 
	 */
	@Override
	public void start() 
		// TODO Auto-generated method stub
		try 
			//发起注册
			String paramStr = GsonUtils.toJson(registParam);
			log.info("SocketConnector Status:,Info:,REGIST:",
					this.socketConnector.isActive(),registParam.getContent(),paramStr);
			this.socketConnector.sendMessage(ByteBuffer.wrap(paramStr.getBytes()));
			//启动音频获取
			curThread = new Thread(this);
			curThread.start();
		 catch (Exception e) 
			log.error("发起注册请求异常", e);
		
	

	@Override
	public void onMessage(ByteBuffer byteBuffer) 
		// TODO Auto-generated method stub
		try 
			if(null == sourceDataLine)
				AudioFormat audioFormat = new AudioFormat(8000, 16, 1, true ,false);
				DataLine.Info info = new DataLine.Info(SourceDataLine.class, audioFormat);
				sourceDataLine = (SourceDataLine) AudioSystem.getLine(info);以上是关于基于Java开发客户端音频采集播放UDP协议转发程序的主要内容，如果未能解决你的问题，请参考以下文章
 

 基于UDP协议的网络通信应用程序（UDP-Socket）


 JAVA NIO udp 实现 群转发


 从 UDP 接收到的消息播放音频，点击声音（Naudio API）


 web音频流转发之音视频直播


 IO 直接播放来自 UDP 流 (NSData) 的原始音频


 Java学习总结（十三）——基于UDP协议网络编程