gb28181协议常见流程简析

Posted 2023-04-27

参考技术A 上级=sip+源设备

1.sip代理或下级向sip服务器发送REGISTER请求+sip编号+IP+端口
2.sip服务器回复401+unauthorized
3.sip代理或下级向sip服务器发送REGISTER请求+sip编号+IP+端口
4.sip服务器回复200

1.以message的方法发送一段xml文本，主要通过Catalog命令和DeviceID发送，获取设备,
2.设备收到后，马上回复200
3.还会回复设备相关信息给服务器
4.服务器收到后回复200，给设备

流程描述如下：
a) 1：媒体流接收者向 SIP 服务器发送 Invite 消息，消息头域中携带 Subject 字段，表明点播的视频源 ID、分辨率、媒体流接收者 ID、接收端媒体流序列号等参数，SDP 消息体中 s 字段为“Play”代表实时点播；
b) 2：SIP 服务器收到 Invite 请求后，通过三方呼叫控制建立媒体服务器和媒体流发送者之间的媒体连接。向媒体服务器发送 Invite 消息，此消息不携带 SDP 消息体；
c) 3：媒体服务器收到 SIP 服务器的 Invite 请求后，回复 200 OK 响应，携带 SDP 消息体，消息体中描述了媒体服务器接收媒体流的 IP、端口、媒体格式等内容；
d) 4：SIP 服务器收到媒体服务器返回的 200 OK 响应后，向媒体流发送者发送 Invite 请求，请求中携带消息 3 中媒体服务器回复的 200 OK 响应消息体，并且修改 s 字段为“Play”代表实时点播，增加 y 字段描述 SSRC 值，f 字段描述媒体参数；
e) 5：媒体流发送者收到 SIP 服务器的 Invite 请求后，回复 200 OK 响应，携带 SDP 消息体，消息体中描述了媒体流发送者发送媒体流的 IP、端口、媒体格式、SSRC 字段等内容；
f) 6：SIP 服务器收到媒体流发送者返回的 200 OK 响应后，向媒体服务器发送ACK 请求，请求中携带消息 5 中媒体流发送者回复的 200 OK 响应消息体，完成与媒体服务器的 Invite 会话建立过程；
g) 7：SIP 服务器收到媒体流发送者返回的 200 OK 响应后，向媒体流发送者发送 ACK 请求，请求中不携带消息体，完成与媒体流发送者的 Invite 会话建立过程；
h) 8：完成三方呼叫控制后，SIP 服务器通过 B2BUA 代理方式建立媒体流接收者和媒体服务器之间的媒体连接。在消息 1 中增加 SSRC 值，转发给媒体服务器；
i) 9：媒体服务器收到 Invite 请求，回复 200 OK 响应，携带 SDP 消息体，消息体中描述了媒体服务器发送媒体流的 IP、端口、媒体格式、SSRC 值等内容；
j) 10：SIP 服务器将消息 9 转发给媒体流接收者；
k) 11：媒体流接收者收到 200 OK 响应后，回复 ACK 消息，完成与 SIP 服务器的 Invite 会话建立过程；
l) 12：SIP 服务器将消息 11 转发给媒体服务器，完成与媒体服务器的 Invite 会话建立过程；
m) 13：媒体流接收者向 SIP 服务器发送 BYE 消息，断开消息 1、10、11 建立的同媒体流接收者的 Invite 会话；
n) 14：SIP 服务器收到 BYE 消息后回复 200 OK 响应，会话断开；
o) 15：SIP 服务器收到 BYE 消息后向媒体服务器发送 BYE 消息，断开消息 8、9、12 建立的同媒体服务器的 Invite 会话；
p) 16：媒体服务器收到 BYE 消息后回复 200 OK 响应，会话断开；
q) 17：SIP 服务器向媒体服务器发送 BYE 消息，断开消息 2、3、6 建立的同媒体服务器的 Invite会话；
r) 18：媒体服务器收到 BYE 消息后回复 200 OK 响应，会话断开；
s) 19：SIP 服务器向媒体流发送者发送 BYE 消息，断开消息 4、5、7 建立的同媒体流发送者的Invite 会话；
t) 20：媒体流发送者收到 BYE 消息后回复 200 OK 响应，会话断开。

message(recordinfo+deviceID+time)->200
设备端xml类型码流发给服务器，发完后以bye结束，服务器回复200

a) 1：源设备向 SIP 服务器发送设备控制命令，设备控制命令采用 MESSAGE 方法携带；
b) 2：SIP 服务器收到命令后返回 200 OK；
c) 3：SIP 服务器向目标设备转发设备控制命令，设备控制命令采用 MESSAGE 方法携带；
d) 4：目标设备收到命令后返回 200 OK；
e) 5：目标设备向 SIP 服务器发送设备控制响应命令，设备控制响应命令采用 MESSAGE 方法携带；
f) 6：SIP 服务器收到命令后返回 200 OK；
g) 7：SIP 服务器向源设备转发设备控制响应命令，设备控制响应命令采用 MESSAGE 方法携带；
h) 8：目标设备收到命令后返回 200 OK

Nacos简析：distro协议

一. CAP理论

• 一致性

• 可用性

• 分区容错性

为什么只能三选二及在分布式场景下P一定不能被放弃？

https://blog.csdn.net/qq_28107929/article/details/106104277

  CAP文氏图 

二. CP的劣势，AP的优势

没有纯粹的CP和AP，保证了CP再尽量靠近A，保证了AP再尽量靠近C。

以RAFT为代表的“强一致性”协议（选主，复制）带来了什么劣势

保障一致性需要做的事情

• 强一致性--->写请求来时日志复制-->主从区分-->选主服务不可用-->投票选主需要一定数量的节点-->可靠性偏低

• 强一致性-->读请求来时请求转发-->主节点压力山大-->性能受限

如果抛弃强一致性我们可以获得什么

• 或许无需选主

• 主节点读写压力分摊

• 不强制要求节点数量即可提供服务

对于服务中心来说，可用，可靠，高性能是第一，反之如果是金融，电商场景，数据一致性则更重要！

三. Distro简介

实现一个AP协议可以参考的对象很少，在nacos-distro出来之前，AP协议代表有以下

• eureka实现的AP协议（去中心化）

• redis-cluster高可用模式（主从复制是异步的）

nacos作为AP派注册中心的粉丝，很多实现上都借鉴了eureka，而Distro作为nacos-ap协议（同时也通过jraft实现raft用作cp，配置中心）的实现，当前仅仅也体现在nacos中开源出来，并没有单独剥离，因此只能说是注册中心的AP协议而非纯粹AP的框架，和etcd的etcd-raft定位很大不同。

distro在nacos中位置

四.Distro部分机制走读

distro部分组件

1. 为什么需要转发，转发到那里，如何转发？

• 每个nacos-server类似分片的数据库，只负责部分实例信息的维护。

• 每个nacos-server类似分片的数据库，只负责部分实例信息的维护。

 private int distroHash(String responsibleTag) { return Math.abs(responsibleTag.hashCode() % Integer.MAX_VALUE); }  

• @CanDistro拦截，调用distroMapper的mapsrv决定转发至哪台nasoc-server，用http形式去转发获取结果。

  简单的distromapper

2.数据初始化，增量同步的基本流程

•  private void load() throws Exception { while (memberManager.allMembersWithoutSelf().isEmpty()) { Loggers.DISTRO.info("[DISTRO-INIT] waiting server list init..."); TimeUnit.SECONDS.sleep(1); } while (distroComponentHolder.getDataStorageTypes().isEmpty()) { Loggers.DISTRO.info("[DISTRO-INIT] waiting distro data storage register..."); TimeUnit.SECONDS.sleep(1); } for (String each : distroComponentHolder.getDataStorageTypes()) { if (!loadCompletedMap.containsKey(each) || !loadCompletedMap.get(each)) { loadCompletedMap.put(each, loadAllDataSnapshotFromRemote(each)); } } }  private boolean loadAllDataSnapshotFromRemote(String resourceType) { DistroTransportAgent transportAgent = distroComponentHolder.findTransportAgent(resourceType); DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType); if (null == transportAgent || null == dataProcessor) { Loggers.DISTRO.warn("[DISTRO-INIT] Can't find component for type {}, transportAgent: {}, dataProcessor: {}", resourceType, transportAgent, dataProcessor); return false; } for (Member each : memberManager.allMembersWithoutSelf()) { try { Loggers.DISTRO.info("[DISTRO-INIT] load snapshot {} from {}", resourceType, each.getAddress()); DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress()); boolean result = dataProcessor.processSnapshot(distroData); Loggers.DISTRO .info("[DISTRO-INIT] load snapshot {} from {} result: {}", resourceType, each.getAddress(), result); if (result) { return true; } } catch (Exception e) { Loggers.DISTRO.error("[DISTRO-INIT] load snapshot {} from {} failed.", resourceType, each.getAddress(), e); } } return false; }  

• 增量同步通常在dataStore被PUT更新之后（注册实例的最后一步就会调用PUT）再主动触发，因为是异步复制，因此这里只是生成一个延迟的任务，塞入DistroTaskEngine的队列中就马上返回

  
  

   public void put(String key, Record value) throws NacosException { onPut(key, value); //... distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE, globalConfig.getTaskDispatchPeriod() / 2); } public void sync(DistroKey distroKey, DataOperation action, long delay) { for (Member each : memberManager.allMembersWithoutSelf()) { DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(), each.getAddress()); DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay); distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask); //... } } } 

• dataStore的结构长什么样子，同步的数据又长什么样，从全量的快照数据到内存数据中经历了那些操作？

dataStore基于concurrentHashMap存放，其结构较为简单为String--->Datum（Key+Record泛型）

同步的数据开始为[]byte，在Serializer中找到实现Record的class，如instance，通过jackson进行数据转化成期望对象，当前默认nacos-server之间同步的全为instance的数据，获取到数据之后，Distro中需要触发对应listeners。

如instance，在dataStore中存放的key为Instance_LIST_PREFIX+namespaceid+serviceName，value为[]instance