快捷導(dǎo)航

Java語言獲取TCP流的實現(xiàn)步驟

更新時間：2023年11月29日 08:30:42 作者：半夏之沫

使用Wireshark分析網(wǎng)絡(luò)包時,一個很常用的功能就是選中一個TCP報文,然后查看這個TCP報文的TCP流,從而可以進一步分析建連是否慢了,斷連是否正常等情況,那么本文就TCP流的概念以及在Java中如何獲取,做一個簡單的學(xué)習(xí),需要的朋友可以參考下

正文

一. TCP流概念

如果去搜索引擎搜索：什么是TCP流，那么大概率是很難得到一個有效的答案的，而在Wireshark中，選中一個TCP報文并單擊右鍵時，在菜單的追蹤流中可以選擇到TCP流這個功能，如下所示。

當(dāng)點擊TCP流后，Wireshark會把選中的TCP報文對應(yīng)的TCP連接的所有TCP報文過濾出來并順序展示，那么這里就知道了，TCP流就是一次TCP連接中，從連接建立，到數(shù)據(jù)傳輸，再到連接斷開整個過程中的TCP報文集合。

那么Wireshark憑什么可以從那么多TCP報文中，精確的把某一條TCP連接的TCP報文過濾出來并順序展示呢，其實就是基于TCP報文的序列號和確認號。下面是TCP報文頭的格式。

可以看到每個TCP報文都有一個序列號SeqNum和確認號AckNum，并且他們的含義如下。

序列號：表示本次傳輸?shù)臄?shù)據(jù)的起始字節(jié)在整個TCP連接傳輸?shù)淖止?jié)流中的編號。舉個例子，某個TCP報文的SeqNum為500，然后報文長度length為100，則表示本次傳輸數(shù)據(jù)的起始字節(jié)在整個TCP流中的序列號為100，并且本次傳輸?shù)臄?shù)據(jù)的序列號范圍是500到599，根據(jù)序列號，能夠?qū)鬏數(shù)臄?shù)據(jù)有序的排列組合起來，以解決網(wǎng)絡(luò)傳輸中的數(shù)據(jù)亂序問題；
確認號：用來告訴對端本端期望下一次收到的數(shù)據(jù)的序列號，換言之，告訴對端本端已經(jīng)正常接收了序列號等于AckNum之前的所有數(shù)據(jù)，根據(jù)確認號，可以解決網(wǎng)絡(luò)傳輸中的數(shù)據(jù)丟包問題。

那么序列號和確認號的變化有什么規(guī)則呢，規(guī)則總結(jié)如下。

本次發(fā)送報文的SeqNum等于上一次發(fā)送報文的SeqNum加上上一次發(fā)送報文的length；
本次發(fā)送報文的AckNum等于上一次接收報文的SeqNum加上上一次接收報文的length；
SYN報文和FIN報文的length默認為1，而不是0。

結(jié)合下面一張圖，可以更好的理解上面的變化規(guī)則。

二. TCP流獲取的Java實現(xiàn)

結(jié)合第一節(jié)的內(nèi)容，想要獲取某一個TCP報文所屬TCP連接的TCP流，其實就可以根據(jù)這個報文的SeqNum和AckNum，向前和向后查找符合序列號和確認號變化規(guī)則的報文，只要符合規(guī)則，那么這個報文就是屬于TCP流的。

在Java語言中，要實現(xiàn)TCP流的獲取，可以先借助io.pkts工具把網(wǎng)絡(luò)包先解開，然后把每個報文封裝為我們自定義的Entity（io.pkts工具包解開后的報文對象不太易用），最后就是根據(jù)序列號和確認號的變化規(guī)則，來得到某一個報文所屬的TCP流。

現(xiàn)在進行實操，先引入io.pkts工具的依賴，如下所示。

<dependency>
    <groupId>io.pkts</groupId>
    <artifactId>pkts-streams</artifactId>
    <version>3.0.10</version>
</dependency>
<dependency>
    <groupId>io.pkts</groupId>
    <artifactId>pkts-core</artifactId>
    <version>3.0.10</version>
</dependency>

同時自定義一個TCP報文的Entity，如下所示。

/**
 * TCP報文Entity。
 */
@Getter
@Setter
@AllArgsConstructor
public class TcpPackage {

    /**
     * 源地址IP。
     */
    private String sourceIp;
    /**
     * 源地址端口。
     */
    private int sourcePort;
    /**
     * 目的地址IP。
     */
    private String destinationIp;
    /**
     * 目的地址端口。
     */
    private int destinationPort;
    /**
     * 報文載荷長度。
     */
    private int length;
    /**
     * ACK報文標(biāo)識。
     */
    private boolean ack;
    /**
     * FIN報文標(biāo)識，
     */
    private boolean fin;
    /**
     * SYN報文標(biāo)識。
     */
    private boolean syn;
    /**
     * RST報文標(biāo)識。
     */
    private boolean rst;
    /**
     * 序列號。
     */
    private long seqNum;
    /**
     * 確認號。
     */
    private long ackNum;
    /**
     * 報文到達時間戳。
     */
    private long arriveTimestamp;
    /**
     * 報文體。
     */
    private String body;

}

現(xiàn)在假設(shè)已經(jīng)拿到了網(wǎng)絡(luò)包對應(yīng)的MultipartFile，下面給出基于io.pkts工具解析網(wǎng)絡(luò)包的實現(xiàn)，如下所示。

public static List<TcpPackage> parseTcpPackagesFromFile(MultipartFile multipartFile) {
    List<TcpPackage> tcpPackages = new ArrayList<>();
    try (InputStream inputStream = multipartFile.getInputStream()) {
        GZIPInputStream gzipInputStream = new GZIPInputStream(inputStream);
        Pcap pcap = Pcap.openStream(gzipInputStream);
        pcap.loop(packet -> {
            if (packet.hasProtocol(Protocol.TCP)) {
                TCPPacket tcpPacket = (TCPPacket) packet.getPacket(Protocol.TCP);
                tcpPackages.add(convertTcpPacket2TcpPackage(tcpPacket));
            }
            return true;
        });
        return tcpPackages;
    } catch (Exception e) {
        String message = "從網(wǎng)絡(luò)包解析TCP報文失敗";
        log.error(message);
        throw new RuntimeException(message, e);
    }
}

上述實現(xiàn)中，假定網(wǎng)絡(luò)包是gzip的壓縮格式，所以使用了GZIPInputStream來包裝網(wǎng)絡(luò)包文件的輸入流，同時因為我們要獲取的是TCP流，所以我們只處理有TCP協(xié)議的報文，并會在convertTcpPacket2TcpPackage() 方法中完成到TcpPackage結(jié)構(gòu)的轉(zhuǎn)換，convertTcpPacket2TcpPackage() 方法實現(xiàn)如下所示。

public static TcpPackage convertTcpPacket2TcpPackage(TCPPacket tcpPacket) {
    // 報文長度=IP報文長度-IP報文頭長度-TCP報文頭長度
    IPPacket ipPacket = tcpPacket.getParentPacket();
    int length = ipPacket.getTotalIPLength() - ipPacket.getHeaderLength() - tcpPacket.getHeaderLength();
    Buffer bodyBuffer = tcpPacket.getPayload();
    String body = ObjectUtils.isNotEmpty(bodyBuffer)
            ? bodyBuffer.toString() : StringUtils.EMPTY;
    long arriveTimestamp = tcpPacket.getArrivalTime() / 1000;
    return new TcpPackage(ipPacket.getSourceIP(), tcpPacket.getSourcePort(), ipPacket.getDestinationIP(), tcpPacket.getDestinationPort(),
            length, tcpPacket.isACK(), tcpPacket.isFIN(), tcpPacket.isSYN(), tcpPacket.isRST(), tcpPacket.getSequenceNumber(),
            tcpPacket.getAcknowledgementNumber(), arriveTimestamp, body);
}

上述方法需要注意的一點就是TCP報文載荷長度的獲取，我們能夠拿到的數(shù)據(jù)是IP報文長度，IP報文頭長度和TCP報文頭長度，所以IP報文長度減去IP報文頭長度可以得到TCP報文長度，再拿TCP報文長度減去TCP報文頭長度就能得到TCP報文載荷長度。

現(xiàn)在我們已經(jīng)拿到網(wǎng)絡(luò)包里面所有TCP報文的集合了，并且這些報文是按照時間先后順序進行正序排序的，我們隨機選中一個報文，拿到這個TCP報文以及其在集合中的索引，然后我們就可以基于下面的實現(xiàn)拿到對應(yīng)的TCP流。

public static List<TcpPackage> getTcpStream(List<TcpPackage> tcpPackages, int index) {
    LinkedList<TcpPackage> tcpStream = new LinkedList<>();
    TcpPackage beginTcpPackage = tcpPackages.get(index);
    long currentSeqNum = beginTcpPackage.getSeqNum();
    long currentAckNum = beginTcpPackage.getAckNum();
    // 從index位置向前查找
    for (int i = index - 1; i >=0; i--) {
        TcpPackage previousTcpPackage = tcpPackages.get(i);
        long previousSeqNum = previousTcpPackage.getSeqNum();
        long previousAckNum = previousTcpPackage.getAckNum();
        if (isPreviousTcpPackageSatisfied(currentSeqNum, currentAckNum, previousSeqNum, previousAckNum)) {
            tcpStream.addFirst(previousTcpPackage);
            currentSeqNum = previousSeqNum;
            currentAckNum = previousAckNum;
        }
    }
    // index位置的報文也要放到tcp流中
    tcpStream.add(beginTcpPackage);
    currentSeqNum = beginTcpPackage.getSeqNum();
    currentAckNum = beginTcpPackage.getAckNum();
    // 從index位置向后查找
    for (int i = index + 1; i < tcpPackages.size(); i++) {
        TcpPackage nextTcpPackage = tcpPackages.get(i);
        long nextSeqNum = nextTcpPackage.getSeqNum();
        long nextAckNum = nextTcpPackage.getAckNum();
        if (isNextTcpPackageSatisfied(currentSeqNum, currentAckNum, nextSeqNum, nextAckNum)) {
            tcpStream.add(nextTcpPackage);
            currentSeqNum = nextSeqNum;
            currentAckNum = nextAckNum;
        }
    }
    return tcpStream;
}

上述方法中，向前查找時判斷TCP報文是否屬于TCP流是基于isPreviousTcpPackageSatisfied() 方法，向后查找時判斷TCP報文是否屬于TCP流是基于isNextTcpPackageSatisfied() 方法，而這兩個方法其實就是把序列號和確認號的變化規(guī)則翻譯成了代碼，如下所示。

public static boolean isPreviousTcpPackageSatisfied(long currentSeqNum, long currentAckNum,
                                                    long previousSeqNum, long previousAckNum) {
    boolean condition1 = currentSeqNum == previousSeqNum && currentSeqNum != 0;
    boolean condition2 = currentAckNum == previousAckNum && currentAckNum != 0;
    boolean condition3 = currentSeqNum == previousAckNum;
    boolean condition4 = currentAckNum - 1 == previousSeqNum;
    return condition1 || condition2 || condition3 || condition4;
}

public static boolean isNextTcpPackageSatisfied(long currentSeqNum, long currentAckNum,
                                                long nextSeqNum, long nextAckNum) {
    boolean condition1 = currentSeqNum == nextSeqNum && currentSeqNum != 0;
    boolean condition2 = currentAckNum == nextAckNum && currentAckNum != 0;
    boolean condition3 = currentAckNum == nextSeqNum;
    boolean condition4 = currentSeqNum + 1 == nextAckNum;
    return condition1 || condition2 || condition3 || condition4;
}

至此，使用Java語言如何從網(wǎng)絡(luò)包中獲得TCP流就介紹完畢。