Java读取Socket行式NMEA数据时的时间戳异常问题

数据源说明 #

外部气象传感器周期性发送基于行的NMEA数据，格式如下：

$WIMWV,357.0,R,5.2,M,A*1A\r\n$WIMWV,123.0,T,5.2,M,A*1A\r\n

数据固定为两行一组（分别标识为R和T），每行以\r\n结尾；每组发送完成后延迟约0.5秒，再发送下一组。通过Putty验证，接收流为每隔0.5秒收到一组两行数据：

$WIMWV,357.0,R,5.2,M,A*1A\r\n
$WIMWV,123.0,T,5.2,M,A*1A\r\n
//time delay ~0.5s
$WIMWV,357.0,R,5.2,M,A*1A\r\n
$WIMWV,123.0,T,5.2,M,A*1A\r\n
//time delay ~0.5s
...

当前Java接收实现 #

Main.java #

public class Main {
    public static void main(String[] args) throws InterruptedException {

        Receiver receiver = new Receiver(args[0], Integer.parseInt(args[1]));
        Thread t = new Thread(receiver);
        t.setDaemon(true);
        t.start();

        while(true) {
            // 主程序其他逻辑
            Thread.sleep(1000);
        }
    }
}

Receiver.java #

package org.example;

import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.time.Instant;
import java.time.temporal.ChronoUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Receiver implements Runnable {

    private final Logger log = LoggerFactory.getLogger(Receiver.class);
    private Socket socket;
    private boolean running = true;

    private String hostname;
    private int port;

    public Receiver(String hostname, int port) {
        this.hostname = hostname;
        this.port = port;
    }

    private String getNanoTime() {
        return String.valueOf(Instant.now().truncatedTo(ChronoUnit.NANOS));
    };

    @Override
    public void run() {

        while (running) {
            try {
                log.info("尝试连接 {}:{}", hostname, port);
                socket = new Socket(hostname, port);
                socket.setTcpNoDelay(true);

                InputStreamReader isr = new InputStreamReader(socket.getInputStream());
                StringBuilder line = new StringBuilder();
                int ch;
                while ((ch = isr.read()) != -1) {
                    if (ch == '\n') {
                        log.info("Time: {}, Message: {}", getNanoTime(), line.toString());
                        // 处理该行数据
                        line = new StringBuilder();
                    } else if (ch != '\r') {  // 忽略回车符
                        line.append((char) ch);
                    }
                }

            } catch (Exception e) {
                log.error("连接 {}:{} 时发生异常:", hostname, port, e);
            } finally {
                try {
                    if (socket != null) socket.close();
                } catch (IOException e) {
                    log.warn("关闭Socket时异常:", e);
                }
            }

            // 重连延迟
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                log.error("重连延迟被打断: {}", e.toString());
            }
        }
        log.warn("TCPLineReceiver 已停止服务 {}:{}", hostname, port);
    }
}

异常现象 #

控制台日志显示，每次收到四行数据且时间戳完全相同，而非预期的每0.5秒两行：

12:40:22.354 INFO o.e.R -- Time: 2026-04-17T12:40:22.354808600Z, Message: $WIMWV,171.8,R,0.3,M,A*2C
12:40:22.354 INFO o.e.R -- Time: 2026-04-17T12:40:22.354808600Z, Message: $WIMWV,175.1,T,0.7,M,A*23
12:40:22.354 INFO o.e.R -- Time: 2026-04-17T12:40:22.354808600Z, Message: $WIMWV,153.7,R,0.2,M,A*22
12:40:22.354 INFO o.e.R -- Time: 2026-04-17T12:40:22.354808600Z, Message: $WIMWV,172.3,T,0.8,M,A*29
12:40:23.359 INFO o.e.R -- Time: 2026-04-17T12:40:23.359554800Z, Message: $WIMWV,156.3,R,0.3,M,A*22
12:40:23.359 INFO o.e.R -- Time: 2026-04-17T12:40:23.359554800Z, Message: $WIMWV,172.3,T,0.9,M,A*28
12:40:23.359 INFO o.e.R -- Time: 2026-04-17T12:40:23.359554800Z, Message: $WIMWV,148.2,R,0.4,M,A*2B
12:40:23.359 INFO o.e.R -- Time: 2026-04-17T12:40:23.359554800Z, Message: $WIMWV,166.5,T,1.0,M,A*23
...

这导致半数数据的时间戳与Putty接收结果存在偏移，无法准确记录每条NMEA语句的实际接收时间。

需求 #

实现与Putty一致的准确接收效果，保留数据发送的时间间隔，为每条NMEA语句打上正确的系统时间戳（数据本身无自带时间戳）。

1. TCP接收缓冲区累积：即使设置了TcpNoDelay禁用Nagle算法（避免TCP层合并发送数据包），操作系统的TCP接收缓冲区仍会累积数据。传感器每0.5秒发送一组两行数据，当缓冲区累积了两组（四行）数据后才会唤醒应用层读取，导致一次性处理多条数据，时间戳均为读取瞬间的时间，而非每条数据实际到达的时间。
2. InputStreamReader内部缓冲：InputStreamReader默认带有8192字节的内部缓冲区，会一次性从TCP缓冲区读取尽可能多的数据到自身缓冲，再逐字符返回给应用。这会导致应用层读取的字符已经是缓冲中的存量数据，而非实时接收的字节，进而导致多条数据的时间戳相同。

1. 替换InputStreamReader，直接使用InputStream读取 #

去掉InputStreamReader的内部缓冲，直接从Socket的InputStream读取字节，减少中间缓冲带来的延迟：

// 修改Receiver中的读取逻辑
InputStream is = socket.getInputStream();
StringBuilder line = new StringBuilder();
int ch;
while ((ch = is.read()) != -1) {
    if (ch == '\n') {
        log.info("Time: {}, Message: {}", getNanoTime(), line.toString());
        line.setLength(0); // 复用StringBuilder，减少对象创建开销
    } else if (ch != '\r') {
        line.append((char) ch);
    }
}

2. 调整TCP接收缓冲区大小 #

设置更小的TCP接收缓冲区，让操作系统更快将数据推送给应用层，减少累积：

// 在创建Socket后添加以下设置
socket.setReceiveBufferSize(512); // 大小可根据单组数据长度调整，建议小于两组数据总长度

注：该值为操作系统建议值，部分系统可能会自动调整为允许的最小/最大值，但设置较小值可有效降低数据累积概率。

3. 使用NIO SocketChannel实现更精细的读取控制 #

采用Java NIO的SocketChannel可以更精确地控制读取时机，避免缓冲导致的批量读取：

// 替换Receiver中的Socket连接与读取逻辑
SocketChannel channel = SocketChannel.open(new InetSocketAddress(hostname, port));
channel.configureBlocking(true); // 保持阻塞模式，与原逻辑一致
ByteBuffer buffer = ByteBuffer.allocate(128); // 缓冲区大小适配单条NMEA语句长度
StringBuilder line = new StringBuilder();

while (running) {
    int bytesRead = channel.read(buffer);
    if (bytesRead == -1) break; // 连接断开

    buffer.flip(); // 切换为读模式
    while (buffer.hasRemaining()) {
        byte b = buffer.get();
        if (b == '\n') {
            log.info("Time: {}, Message: {}", getNanoTime(), line.toString());
            line.setLength(0);
        } else if (b != '\r') {
            line.append((char) b);
        }
    }
    buffer.clear(); // 切换为写模式，准备下次读取
}

这种方式可以实时处理每个到达的字节，最大程度贴近数据实际接收时间。

关键注意事项 #

• 时间戳记录时机：必须在检测到换行符的瞬间调用getNanoTime()，确保时间戳是数据实际接收完成的时间，而非从缓冲中读取的时间。
• 缓冲禁用优先级：优先去掉InputStreamReader这类应用层缓冲，这是导致批量读取的主要原因。
• TcpNoDelay保持启用：当前代码中socket.setTcpNoDelay(true)的设置是正确的，需保留以避免TCP层合并小数据包。

内容的提问来源于stack exchange，提问作者Alex