My Technical sharing

0%

Android上利用Ntp实现不同设备间的同步

Posted on Edited on In

最近公司有个需求，希望在不同的Android设备上实现视频的播放同步，误差尽可能的小，后期继续优化实现逐帧同步效果，以便后期进行矩阵拼接屏的研发，前期处于试探性阶段，在App上实现，目前误差在10-20ms内，属于肉眼完全看不出的误差效果范围，只有利用手机的慢动作摄影才能看出一点过渡误差，并且视频播放出的声音听起来也完全同步。对于请求主机时间消息的传递误差，主要利用了ntp时间同步技术来解决，对于视频的加载耗时通过两个播放器实例交替加载，达到预先加载的效果来解决，加载视频耗时造成的误差基本可以忽略不计。只要获取到了当前的主机时间，那么就能推算出主机的播放时间，这个时间可以看作是所有设备同步播放的时间，也能算出当前设备时间到这个同步播放时间的偏移量，达到准时播放同一个视频的效果。由于设备不一定一直联网，排除设备的时钟时间错误造成的干扰，这里统一采用rtc时间（硬件时钟）。这里主要记录一下在Android平台上ntp协议的应用。

NTP协议简介

网络时间协议NTP(Network Time Protocol)用于将计算机客户或服务器的时间与另一服务器同步，使用层次式时间分布模型。在配置时，NTP可以利用冗余服务器和多条网络路径来获得时间的高准确性和高可靠性。即使客户机在长时间无法与某一时间服务器相联系的情况下,仍可提供高准确度时间。

联网计算机同步时钟最简便的方法是网络授时。网络授时分为广域网授时和局域网授时。广域网授时精度通常能达50ms级，但有时超过500ms，这是因为每次经过的路由器路径可能不相同。现在还没有更好的办法将这种不同路径延迟的时间误差完全消除。局域网授时不存在路由器路径延迟问题，因而授时精度理论上可以提到亚毫秒级。Windows内置NTP服务，在局域网内其最高授时精度也只能达10ms级。

进一步提高NTP授时精度的方法

局域网络延相对较大的原因在于时间戳的发送请求和接收一般都是在应用层。减少操作系统内核处理延时可以提高NTP授时精度，使发/收NTP包时间戳应尽量接近主机真实发/收包时刻。在不改变硬件的条件下，修改网卡驱动程序，将记录NTP包发/收时间戳从应用程序移至网卡驱动程序处，可消除操作系统内核处理延时不确定而造成的误差。这种方法在局域网中可大幅提高NTP授时精度至μs级。

NTP的算法推导

NTP最典型的授时方式是Client/Server方式。如下图所示，客户机首先向服务器发送一个NTP包，其中包含了该包离开客户机的时间戳T1，当服务器接收到该包时，依次填入包到达的时间戳T2、包离开的时间戳T3，然后立即把包返回给客户机。客户机在接收到响应包时，记录包返回的时间戳T4。客户机用上述4个时间参数就能够计算出2个关键参数：NTP包的往返延迟d和客户机与服务器之间的时钟偏差t。客户机使用时钟偏差来调整本地时钟，以使其时间与服务器时间一致。

T1为客户发送NTP请求时间戳(以客户时间为参照)；T2为服务器收到NTP请求时间戳(以服务器时间为参照)；T3为服务器回复NTP请求时间戳(以服务器时间为参照)；T4为客户收到NTP回复包时间戳(以客户时间为参照)；d1为NTP请求包传送延时，d2为NTP回复包传送延时；t为服务器和客户端之间的时间偏差，d为NTP包的往返时间
那么服务器的处理时间为T3 - T2，服务器从发送请求到接收应答总时间为T4 - T1
例如T1客户发送NTP请求时间为10:00:00am，T2服务器收到NTP请求时间为11:00:01am，T3服务器回复NTP请求时间为11:00:02am，T4客户端收到NTP回复包10:00:03am，那么
d = (T4 - T1) - (T3 - T2) = 2秒
客户端设备相对服务端设备的时间差为
t = (T2 - T1) + (T3 - T4) / 2 约等于 1 小时

Android代码实现

客户端请求ntp时间

已知客户端T1，T4时间，实现从服务器返回的报文中获取T2，T3时间参数并计算服务器ntp时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
   private static final int NTP_PACKET_SIZE = 48;
   private static final int NTP_MODE_CLIENT = 3;
   private static final int NTP_VERSION = 3;
   private static final int RECEIVE_TIME_OFFSET = 32;
   private static final int TRANSMIT_TIME_OFFSET = 40;
   private static final long OFFSET_1900_TO_1970 = ((365L * 70L) + 17L) * 24L * 60L * 60L;
/**
    * 请求ntp服务器设备的时间，设置服务器IP地址，端口号，以及超时时间
    *
    * @param ipAddress ip地址
    * @param port      端口号
    * @param timeout   超时时间单位毫秒
    * @return 如果请求成功返回true
    */
   private boolean requestTime(String ipAddress, int port, int timeout) {
       try {
           InetAddress address = InetAddress.getByName(ipAddress);
           DatagramSocket mSocket = new DatagramSocket();
           mSocket.setSoTimeout(timeout);
           byte[] buffer = new byte[NTP_PACKET_SIZE];
           DatagramPacket request = new DatagramPacket(buffer, buffer.length, address, port);
           buffer[0] = NTP_MODE_CLIENT | (NTP_VERSION << 3);
           // 请求发送时间
           final long requestTicks = SystemClock.elapsedRealtime();
           mSocket.send(request);
           // 读取服务器响应数据包
           DatagramPacket response = new DatagramPacket(buffer, buffer.length);
           mSocket.receive(response);
           // 收到服务器响应时间
           final long responseTicks = SystemClock.elapsedRealtime();
           // 服务器的接收时间
           final long receiveTime = readTimeStamp(buffer, RECEIVE_TIME_OFFSET);
           // 服务器的发送时间
           final long transmitTime = readTimeStamp(buffer, TRANSMIT_TIME_OFFSET);
           // 计算时间客户端和服务器的时间差（服务器收到请求时间 - 客户端请求时间） + （服务器回复请求时间 - 客户端收到回复时间） / 2
           long clockOffset = ((receiveTime - requestTicks) + (transmitTime - responseTicks)) / 2;
           // 当前ntp服务器的时间
           long mNtpTime = SystemClock.elapsedRealtime() + clockOffset;
           // 添加到平均值计算集合中，求平均值使得计算误差更小
           addDeviation(clockOffset);
       } catch (Exception e) {
           return false;
       } finally {
           try {
               if (mSocket != null) {
                   if (!mSocket.isClosed()) {
                       mSocket.close();
                   }
                   mSocket.disconnect();
                   mSocket = null;
               }
           } catch (Exception e) {
               e.printStackTrace();
           }
       }
       return true;
   }

/**
    * 从buffer中读取32位的大端数
    */
   private long read32(byte[] buffer, int offset) {
       byte b0 = buffer[offset];
       byte b1 = buffer[offset + 1];
       byte b2 = buffer[offset + 2];
       byte b3 = buffer[offset + 3];
       // convert signed bytes to unsigned values
       int i0 = ((b0 & 0x80) == 0x80 ? (b0 & 0x7F) + 0x80 : b0);
       int i1 = ((b1 & 0x80) == 0x80 ? (b1 & 0x7F) + 0x80 : b1);
       int i2 = ((b2 & 0x80) == 0x80 ? (b2 & 0x7F) + 0x80 : b2);
       int i3 = ((b3 & 0x80) == 0x80 ? (b3 & 0x7F) + 0x80 : b3);

       return ((long) i0 << 24) + ((long) i1 << 16) + ((long) i2 << 8)
               + (long) i3;
   }

   /**
    * 从buffer中读取时间戳
    * 返回毫秒数
    */
   private long readTimeStamp(byte[] buffer, int offset) {
       // 读取秒数
       long seconds = read32(buffer, offset);
       // 读取小数
       long fraction = read32(buffer, offset + 4);
       return ((seconds - OFFSET_1900_TO_1970) * 1000)
               + ((fraction * 1000L) / 0x100000000L);
   }

这里的SystemClock.elapsedRealtime()是获取设备启动后到现在的时间（硬件时间），System.currentTimeMillis()获取的是系统时间（软件时间），是距离1970年1月1日开始计算的一个值，这里为了减少误差，避免因为外界修改软件时间而导致时间不准的情况而使用硬件时间。

服务器返回ntp时间

从客户端发的报文中获取响应地址和端口号，把收到请求的时间写入报文并保存到链表，从链表中取出刚才放入的报文，写入发送响应的时间发给客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
private static final long OFFSET_1900_TO_1970 = ((365L * 70L) + 17L) * 24L * 60L * 60L;
private static final int RECEIVE_TIME_OFFSET = 32;
private static final int TRANSMIT_TIME_OFFSET = 40;
private static final int NTP_PACKET_SIZE = 48;
private static final int NTP_MODE_SERVIER = 0;
private static final int NTP_VERSION = 3;
public void start(int port) {
    mCompositeDisposable.add(Observable.just(1).subscribeOn(Schedulers.io())
            .subscribe(integer -> {
                start = true;
                Logger.t(TAG).d("服务端开启 port=" + port);
                try {
                    mSocket = new DatagramSocket(null);
                    // 设置端口重用
                    mSocket.setReuseAddress(true);
                    // 绑定端口号
                    mSocket.bind(new InetSocketAddress(port));
                    while (start) {
                        try {
                            byte[] buffer = new byte[NTP_PACKET_SIZE];
                            DatagramPacket response = new DatagramPacket(buffer,
                                    buffer.length);
                            buffer[0] = NTP_MODE_SERVIER | (NTP_VERSION << 3);
                            Logger.t(TAG).d("等待接收客户端消息");
                            mSocket.receive(response);
                            DatagramPacket echo = new DatagramPacket(buffer, buffer.length, response.getAddress(), response.getPort());
                            Logger.t(TAG).d("收到客户端请求=" + response.getAddress());
                            // 把收到的时间写入buffer
                            writeTimeStamp(buffer, RECEIVE_TIME_OFFSET, SystemClock.elapsedRealtime());
                            // 添加到链表
                            mLinkedList.add(echo);
                        } catch (IOException e) {
                            Logger.t(TAG).e("while receive failed:" + e);
                        }
                    }
                } catch (Exception e) {
                    Logger.t(TAG).e("response failed:" + e);
                } finally {
                    Logger.t(TAG).e("关闭socket");
                    if (mSocket != null) {
                        if (!mSocket.isClosed()) {
                            mSocket.close();
                        }
                        mSocket.disconnect();
                        mSocket = null;
                    }
                }
            }));
    mCompositeDisposable.add(Observable.just(1).subscribeOn(Schedulers.io())
            .subscribe(integer -> {
                mSendSocket = new DatagramSocket();
                mSendSocket.setSoTimeout(2000);
                while (start) {
                    try {
                        if (mLinkedList.size() > 0) {
                            // 取出刚才放入链表的数据包
                            DatagramPacket first = mLinkedList.removeFirst();
                            if (first != null) {
                                // 写入发送响应包的时间
                                writeTimeStamp(first.getData(), TRANSMIT_TIME_OFFSET, SystemClock.elapsedRealtime());
                                Logger.t(TAG).d("向客户端发送消息=" + first.getAddress());
                                mSendSocket.send(first);
                            }
                        }
                    } catch (IOException e) {
                        Logger.t(TAG).e("while send failed:" + e);
                    }
                }
                Logger.t(TAG).d("发送消息端关闭");
            }));
}

/**
 * 写入硬件时间到buffer中
 */
private void writeTimeStamp(byte[] buffer, int offset, long time) {
    long seconds = time / 1000L;
    long milliseconds = time - seconds * 1000L;
    seconds += OFFSET_1900_TO_1970;

    // 按大端模式写入秒数
    buffer[offset++] = (byte) (seconds >> 24);
    buffer[offset++] = (byte) (seconds >> 16);
    buffer[offset++] = (byte) (seconds >> 8);
    buffer[offset++] = (byte) (seconds >> 0);

    long fraction = milliseconds * 0x100000000L / 1000L;
    // 按大端模式写入小数
    buffer[offset++] = (byte) (fraction >> 24);
    buffer[offset++] = (byte) (fraction >> 16);
    buffer[offset++] = (byte) (fraction >> 8);
    // 低位字节随机
    buffer[offset++] = (byte) (Math.random() * 255.0);
}