Janus架构以及基本开发

2024-01-23 07:08
文章标签 开发 基本 架构 janus

本文主要是介绍Janus架构以及基本开发，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Janus架构以及基本开发

    • 1 前言
    • 2 模块结构
      • 2.1 媒体模块
      • 2.2 信令模块
      • 2.3 插件模块
        • 2.3.1 接口
        • 2.3.2 核心数据结构
    • 3 信令协议
      • 3.1 基本格式
      • 3.2 基本交互流程
    • 4 内部数据流
      • 4.1 接收
      • 4.1.1 信令
        • 4.1.1.1 注册Transport插件
        • 4.1.1.2 创建信令处理线程/线程池
        • 4.1.1.3 分发信令
      • 4.1.2 媒体
        • 4.1.2.1 创建媒体处理线程/线程池
        • 4.1.2.2 设置本地ICE数据回调
        • 4.1.2.3 传递RTP
      • 4.2 发送
        • 4.2.1 信令
        • 4.2.2 媒体
    • 5 开发

1 前言

以前在电信行业做视频会议一般都使用MCU(Multipoint Control Unit)，也就是多方推流在MCU上进行合流，在拉流的时候只有一路合流，这样的好处是无论几方推流，拉流只有一路，下行带宽比较小。但是问题也比较多，只要推流方一多，MCU的压力就比较大，而且分布式的部署也比较难，成本又很高。

现在互联网行业比较流行的是SFU(Selective Forwarding Unit)，简单说就是只负责转发流，不负责合流，压力很小。这样的模式可以依托CDN进行分布式的部署，不过拉流的方数限于客户端的带宽和处理能力。目前比较常见的支持WebRTC的SFU有Janus、Licode、Kurento、Jitsi等。

2 模块结构

本文介绍的Janus版本为0.6.0。

在这里插入图片描述
上图是Janus主要的模块结构，有一些通用工具模块这里没有列出。

2.1 媒体模块

Janus不是简单转发WebRTC的媒体流，还有一定的控制能力，因此需要支持WebRTC的媒体能力，其媒体功能包含以下基本模块：

  • ICE：打洞，负责与Peer的连通，Janus可以部署在NAT后面，使用了libnice；
  • DTLS：UDP版的TLS，就是加密的UDP，WebRTC用来传递SRTP的密钥，使用了OpenSSL/BoringSSL；
  • RTP/RTCP：提供RTP/RTCP封/解包的接口，需要发送一些WebRTC支持的RTCP包，例如FIR、PLI、RR等；
  • SRTP：加密的RTP，开启后WebRTC传输的RTP负载都是加密的；
  • SDP：提供SDP封/解包的接口，用于协商媒体的协议，可以用SDP对WebRTC的一些功能进行设定；
  • SCTP：WebRTC的数据通道使用的协议，就是加上了流控的UDP，可以传输任意数据。

2.2 信令模块

除了媒体协议，Janus还要提供信令交互的协议，传统的信令协议有SIP、XMPP等，Janus上的信令应用协议可以定制，底层主要的传输协议有

  • HTTP(s)；
  • WebSocket(s)；
  • MQTT；
  • NanoMsg；
  • RabbitMQ。

其中WebSocket使用了libwebsockets，HTTP使用了libmicrohttpd。

2.3 插件模块

2.3.1 接口

插件模块是Janus的应用层，可以基于媒体模块、信令模块等底层的接口开发不同的应用。每个插件都被编译成动态库，在Janus启动的时候动态加载。

Janus定义了插件的通用接口，每个插件都必须实现这些接口：

struct janus_plugin {//初始化插件int (* const init)(janus_callbacks *callback, const char *config_path);//销毁插件void (* const destroy)(void);//创建插件上的一个自定义会话void (* const create_session)(janus_plugin_session *handle, int *error);//销毁插件自定义会话void (* const destroy_session)(janus_plugin_session *handle, int *error);//处理信令struct janus_plugin_result * (* const handle_message)(janus_plugin_session *handle, char *transaction, json_t *message, json_t *jsep);//建立媒体，在ICE通道建立后被调用void (* const setup_media)(janus_plugin_session *handle);//处理rtp，可以在这里进行转发void (* const incoming_rtp)(janus_plugin_session *handle, int video, char *buf, int len);//处理rtcp，可以在这里进行转发void (* const incoming_rtcp)(janus_plugin_session *handle, int video, char *buf, int len);//处理数据通道的数据，可以在这里进行转发void (* const incoming_data)(janus_plugin_session *handle, char *buf, int len);//链路恶化通知，可以在这里进行转发void (* const slow_link)(janus_plugin_session *handle, int uplink, int video);//挂断void (* const hangup_media)(janus_plugin_session *handle);	
};

在初始化时插件时，Janus会传入插件一些回调，用于调用数据转发等相关功能：

struct janus_callbacks {//推送一个信令给客户端，可以携带sdpint (* const push_event)(janus_plugin_session *handle, janus_plugin *plugin, const char *transaction, json_t *message, json_t *jsep);//发送rtp给客户端void (* const relay_rtp)(janus_plugin_session *handle, int video, char *buf, int len);//发送rtcp给客户端void (* const relay_rtcp)(janus_plugin_session *handle, int video, char *buf, int len);//发送数据给客户端void (* const relay_data)(janus_plugin_session *handle, char *buf, int len);//关闭WebRTC PeerConnectionvoid (* const close_pc)(janus_plugin_session *handle);//关闭一个会话void (* const end_session)(janus_plugin_session *handle);
};
2.3.2 核心数据结构
  • janus_session：标识Janus上的一个会话；
typedef struct janus_session {guint64 session_id;      //会话IDGHashTable *ice_handles; //janus_ice_handle表……
}
  • janus_ice_handle：标识一个ICE会话，封装了ICE的传输信息，用于一个通道实际的数据收发，一个janus_session上可以有若干janus_ice_handle；
struct janus_ice_handle {void *session;            //从属的Janus会话guint64 handle_id;        //ICE会话句柄void *app;                //绑定的插件void *app_handle;         //插件自定义会话句柄janus_ice_stream *stream; //ICE流NiceAgent *agent;         //nice代理	……
}
  • janus_plugin_session：标识一个底层插件会话，用于绑定ICE会话和插件自定义会话；
struct janus_plugin_session {void *gateway_handle;   //ICE会话句柄void *plugin_handle;    //插件自定义会话句柄
};
  • 插件自定义会话：标识插件内的一个会话。
struct custom_plugin_session {janus_plugin_session *handle; //底层插件会话句柄
};

在这里插入图片描述

3 信令协议

3.1 基本格式

Janus使用JSON传输信令，典型的消息如下：

创建Janus会话命令：
{ "janus": "create", "transaction": "8u8NAbBdgVN1" 
}
响应：
{"janus": "success","transaction": "8u8NAbBdgVN1","data": {"id": 6282808472350684}
}

"janus"后面的字符串代表消息的类型，主要有：

  • create：创建一个Janus会话命令；
  • attach：attach一个插件到Janus会话命令；
  • success：一个命令的成功结果；
  • error：一个命令的失败结果；
  • ack：一个命令的ack，因为不能直接返回结果，先回ack，后续的结果通过event返回；
  • event：推送给客户端的异步事件，主要由插件发出，这些事件需要插件来自己定义；
  • message：客户端发给插件的消息，message和event就构成了应用协议；
  • trickle：客户端发送的candidate，会传递给ICE句柄；
  • keepalive：客户端发送的心跳；
  • webrtcup：ICE成功建立通道的通知；
  • media：音频、视频媒体的接收通知；
  • slowlink：链路恶化通知；
  • hangup：挂断通知；
  • detached：插件从Janus会话detach的通知，释放了一个插件句柄。

"transaction"字段为事务ID，事务的概念可以参考SIP协议，可以理解为一问一答这样一个交互的过程，用事务ID来标识。

3.2 基本交互流程

所有插件都遵循的基本数据流程：

  1. 客户端发送create创建一个Janus会话；
  2. Janus回复success返回Janus会话句柄；
  3. 客户端发送attach命令在Janus会话上attach指定插件；
  4. Janus回复success返回插件的句柄；
  5. 客户端给指定的插件发送message进行信令控制；
  6. Janus上的插件发送event通知事件给客户端；
  7. 客户端收集candidate并通过trickle消息发送给插件绑定的ICE通道；
  8. Janus发送webrtcup通知ICE通道建立；
  9. 客户端发送媒体数据；
  10. Janus发送media消息通知媒体数据的第一次到达；
  11. Janus进行媒体数据转发。

只有粗体部分是插件相关，其他是公用的部分。

4 内部数据流

4.1 接收

4.1.1 信令

4.1.1.1 注册Transport插件

在janus.c中

定义transport回调，所有的transport插件都是用同一个回调：
static janus_transport_callbacks janus_handler_transport = {.incoming_request = janus_transport_incoming_request,.transport_gone = janus_transport_gone,.is_api_secret_needed = janus_transport_is_api_secret_needed,.is_api_secret_valid = janus_transport_is_api_secret_valid,.is_auth_token_needed = janus_transport_is_auth_token_needed,.is_auth_token_valid = janus_transport_is_auth_token_valid,.events_is_enabled = janus_events_is_enabled,.notify_event = janus_transport_notify_event,
};

在main函数中，加载所有transport插件，并注册回调：

char transportpath[1024];
while((transportent = readdir(dir))) {void *transport = dlopen(transportpath, RTLD_NOW | RTLD_GLOBAL);janus_transport->init(&janus_handler_transport, configs_folder)……
}

这样无论底层是什么协议(WebSocket、HTTP)，信令消息都会通过回调janus_transport_incoming_request得到。

4.1.1.2 创建信令处理线程/线程池

在main函数中：

//创建信令请求队列
requests = g_async_queue_new_full((GDestroyNotify) janus_request_destroy);
//创建信令处理/分发线程
GThread *requests_thread = g_thread_try_new("sessions requests", &janus_transport_requests, NULL, &error);
//创建异步信令(给插件的“message”消息)处理线程池
tasks = g_thread_pool_new(janus_transport_task, NULL, -1, FALSE, &error);
4.1.1.3 分发信令

在janus.c中：

void janus_transport_incoming_request(janus_transport *plugin, janus_transport_session *transport, void *request_id, gboolean admin, json_t *message, json_error_t *error) {//信令入队janus_request *request = janus_request_new(plugin, transport, request_id, admin, message);g_async_queue_push(requests, request);
}

static void *janus_transport_requests(void *data) {janus_request *request = NULL;while(!g_atomic_int_get(&stop)) {//从信令请求队列弹出一个信令request = g_async_queue_pop(requests); json_t *message = json_object_get(request->message, "janus");const gchar *message_text = json_string_value(message);//如果是给插件的“message”消息if(message_text && !strcasecmp(message_text, "message")) {//分发到tasks线程池处理GError *tperror = NULL;			g_thread_pool_push(tasks, request, &tperror); } else {//如果不是给插件的“message”消息，则自己处理janus_process_incoming_request(request); }}JANUS_LOG(LOG_INFO, "Leaving Janus requests handler thread\n");return NULL;
}

4.1.2 媒体

4.1.2.1 创建媒体处理线程/线程池

当前版本Janus提供两种线程模式处理媒体，由janus.jcfg的【general/event_loops】配置项决定：

  • 0：对每个媒体会话创建一个线程；
  • n：预先创建包含n个线程的线程池。
    在janus.c的main函数中：
	item = janus_config_get(config, config_general, janus_config_type_item, "event_loops");if(item && item->value)janus_ice_set_static_event_loops(atoi(item->value)); //创建包含event_loops个线程的线程池

在ice.c中：

void janus_ice_set_static_event_loops(int loops) {int i = 0;for(i=0; i<loops; i++) { //创建指定线程janus_ice_static_event_loop *loop = g_malloc0(sizeof(janus_ice_static_event_loop));loop->id = static_event_loops;loop->mainctx = g_main_context_new();loop->mainloop = g_main_loop_new(loop->mainctx, FALSE);GError *error = NULL;char tname[16];g_snprintf(tname, sizeof(tname), "hloop %d", loop->id);loop->thread = g_thread_try_new(tname, &janus_ice_static_event_loop_thread, loop, &error);event_loops = g_slist_append(event_loops, loop);static_event_loops++;}current_loop = event_loops;
}static void *janus_ice_static_event_loop_thread(void *data) {janus_ice_static_event_loop *loop = data;JANUS_LOG(LOG_DBG, "[loop#%d] Looping...\n", loop->id);g_main_loop_run(loop->mainloop); //在线程中运行loop。g_main_loop_unref(loop->mainloop);g_main_context_unref(loop->mainctx);return NULL;
}

在Janus收到“attach”命令后，调用以下函数处理：

gint janus_ice_handle_attach_plugin(void *core_session, janus_ice_handle *handle, janus_plugin *plugin) {if(static_event_loops == 0) {//如果event_loops配置项为0，创建一个新的loop。handle->mainctx = g_main_context_new();handle->mainloop = g_main_loop_new(handle->mainctx, FALSE); //绑定loop到handle} else {//如果使用线程池，则获取线程池中一个线程的loop。janus_refcount_increase(&handle->ref);janus_mutex_lock(&event_loops_mutex);janus_ice_static_event_loop *loop = (janus_ice_static_event_loop *)current_loop->data;handle->mainctx = loop->mainctx;handle->mainloop = loop->mainloop; //绑定loop到handlecurrent_loop = current_loop->next;if(current_loop == NULL)current_loop = event_loops;janus_mutex_unlock(&event_loops_mutex);}handle->rtp_source = janus_ice_outgoing_traffic_create(handle, (GDestroyNotify)g_free);g_source_set_priority(handle->rtp_source, G_PRIORITY_DEFAULT);g_source_attach(handle->rtp_source, handle->mainctx);if(static_event_loops == 0) {//如果event_loops配置项为0，创建一个新的线程。GError *terror = NULL;char tname[16];g_snprintf(tname, sizeof(tname), "hloop %"SCNu64, handle->handle_id);janus_refcount_increase(&handle->ref);handle->thread = g_thread_try_new(tname, &janus_ice_handle_thread, handle, &terror);}
}static void *janus_ice_handle_thread(void *data) {janus_ice_handle *handle = data;g_main_loop_run(handle->mainloop); //在新创建的线程中运行loop。return NULL;
}
4.1.2.2 设置本地ICE数据回调

在收到客户端的信令中携带的sdp(收到offer或者answer)后，Janus会调用以下方法启动ICE：

int janus_ice_setup_local(janus_ice_handle *handle, int offer, int audio, int video, int data, int trickle) {//设置libnice回调参数janus_ice_stream *stream = g_malloc0(sizeof(janus_ice_stream));stream->handle = handle;janus_ice_component *component = g_malloc0(sizeof(janus_ice_component));component->stream = stream;//设置libnice回调，传入component作为返回参数。nice_agent_attach_recv(handle->agent, handle->stream_id, 1, g_main_loop_get_context(handle->mainloop),janus_ice_cb_nice_recv, component);
}
4.1.2.3 传递RTP
static void janus_ice_cb_nice_recv(NiceAgent *agent, guint stream_id, guint component_id, guint len, gchar *buf, gpointer ice) {//得到传入的componentjanus_ice_component *component = (janus_ice_component *)ice;//获取插件指针janus_ice_stream *stream = component->stream;janus_ice_handle *handle = stream->handle;janus_plugin *plugin = (janus_plugin *)handle->app;if(plugin && plugin->incoming_rtp && handle->app_handle &&!g_atomic_int_get(&handle->app_handle->stopped) &&!g_atomic_int_get(&handle->destroyed))//传递rtp给插件plugin->incoming_rtp(handle->app_handle, video, buf, buflen); 
}

4.2 发送

4.2.1 信令

在2.3.1节中说到在插件初始化的时候传入一个janus_callbacks类型的回调，其回调方法push_event可以向客户端发送一个event：

int (*const push_event)(janus_plugin_session *handle, janus_plugin *plugin, const char *transaction, json_t *message, json_t *jsep);

参数：

  • handle：底层插件会话句柄，已经与插件自定义会话句柄绑定，任何时候都能够获取；
  • plugin：插件指针，每个插件维护的一个静态全局指针，任何时候都能够获取；
  • transaction：事务ID，如果是对之前客户端某个命令返回的事件，则应该携带该命令的transaction，否则可以自行指定；
  • message：事件的消息内容；
  • jsep：事件携带的sdp。
4.2.2 媒体

janus_callbacks回调的relay_rtp方法可以向客户端发送rtp：

void (*const relay_rtp)(janus_plugin_session *handle, int video, char *buf, int len);

参数：

  • handle：底层插件会话句柄，已经与插件自定义会话句柄绑定，任何时候都能够获取；
  • video：视频或者音频；
  • buf/len：RTP数据。

5 开发

开发方式不外乎以下两种：

  • 修改默认插件，找到功能比较接近的插件，然后在上面开发可以快速验证；
  • 添加新的插件，需要修改编译配置文件。

主要的开发工作是在插件上设计信令协议，以及RTP包的处理。

这篇关于Janus架构以及基本开发的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/635598。

相关文章

Python Faker库基本用法详解

Python Faker库基本用法详解

《PythonFaker库基本用法详解》Faker是一个非常强大的库,适用于生成各种类型的伪随机数据,可以帮助开发者在测试、数据生成、或其他需要随机数据的场景中提高效率,本文给大家介绍PythonF... 目录安装基本用法主要功能示例代码语言和地区生成多条假数据自定义字段小结Faker 是一个 python
阅读更多...
Python通过模块化开发优化代码的技巧分享

Python通过模块化开发优化代码的技巧分享

《Python通过模块化开发优化代码的技巧分享》模块化开发就是把代码拆成一个个“零件”,该封装封装,该拆分拆分,下面小编就来和大家简单聊聊python如何用模块化开发进行代码优化吧... 目录什么是模块化开发如何拆分代码改进版：拆分成模块让模块更强大：使用 __init__.py你一定会遇到的问题模www.
阅读更多...
Spring Security基于数据库的ABAC属性权限模型实战开发教程

Spring Security基于数据库的ABAC属性权限模型实战开发教程

《SpringSecurity基于数据库的ABAC属性权限模型实战开发教程》：本文主要介绍SpringSecurity基于数据库的ABAC属性权限模型实战开发教程,本文给大家介绍的非常详细,对大... 目录1. 前言2. 权限决策依据RBACABAC综合对比3. 数据库表结构说明4. 实战开始5. MyBA
阅读更多...
使用Python开发一个简单的本地图片服务器

使用Python开发一个简单的本地图片服务器

《使用Python开发一个简单的本地图片服务器》本文介绍了如何结合wxPython构建的图形用户界面GUI和Python内建的Web服务器功能,在本地网络中搭建一个私人的,即开即用的网页相册,文中的示... 目录项目目标核心技术栈代码深度解析完整代码工作流程主要功能与优势潜在改进与思考运行结果总结你是否曾经
阅读更多...
用js控制视频播放进度基本示例代码

用js控制视频播放进度基本示例代码

《用js控制视频播放进度基本示例代码》写前端的时候,很多的时候是需要支持要网页视频播放的功能,下面这篇文章主要给大家介绍了关于用js控制视频播放进度的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言html部分:JavaScript部分:注意：总结前言在javascript中控制视频播放
阅读更多...
Spring Boot + MyBatis Plus 高效开发实战从入门到进阶优化(推荐)

Spring Boot + MyBatis Plus 高效开发实战从入门到进阶优化(推荐)

《SpringBoot+MyBatisPlus高效开发实战从入门到进阶优化(推荐)》本文将详细介绍SpringBoot+MyBatisPlus的完整开发流程,并深入剖析分页查询、批量操作、动... 目录Spring Boot + MyBATis Plus 高效开发实战：从入门到进阶优化1. MyBatis
阅读更多...
Python基于wxPython和FFmpeg开发一个视频标签工具

Python基于wxPython和FFmpeg开发一个视频标签工具

《Python基于wxPython和FFmpeg开发一个视频标签工具》在当今数字媒体时代,视频内容的管理和标记变得越来越重要,无论是研究人员需要对实验视频进行时间点标记,还是个人用户希望对家庭视频进行... 目录引言1. 应用概述2. 技术栈分析2.1 核心库和模块2.2 wxpython作为GUI选择的优
阅读更多...
Java异常架构Exception(异常)详解

Java异常架构Exception(异常)详解

《Java异常架构Exception(异常)详解》：本文主要介绍Java异常架构Exception(异常),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1. Exception 类的概述Exception的分类2. 受检异常（Checked Exception）
阅读更多...
利用Python开发Markdown表格结构转换为Excel工具

利用Python开发Markdown表格结构转换为Excel工具

《利用Python开发Markdown表格结构转换为Excel工具》在数据管理和文档编写过程中,我们经常使用Markdown来记录表格数据,但它没有Excel使用方便,所以本文将使用Python编写一... 目录1.完整代码2. 项目概述3. 代码解析3.1 依赖库3.2 GUI 设计3.3 解析 Mark
阅读更多...
利用Go语言开发文件操作工具轻松处理所有文件

利用Go语言开发文件操作工具轻松处理所有文件

《利用Go语言开发文件操作工具轻松处理所有文件》在后端开发中,文件操作是一个非常常见但又容易出错的场景,本文小编要向大家介绍一个强大的Go语言文件操作工具库,它能帮你轻松处理各种文件操作场景... 目录为什么需要这个工具？核心功能详解1. 文件/目录存javascript在性检查2. 批量创建目录3. 文件
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2