Java语言获取TCP流的实现步骤

<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。
 */
@Getter
@Setter
@AllArgsConstructor
public class TcpPackage {

    /**
     * 源地址IP。
     */
    private String sourceIp;
    /**
     * 源地址端口。
     */
    private int sourcePort;
    /**
     * 目的地址IP。
     */
    private String destinationIp;
    /**
     * 目的地址端口。
     */
    private int destinationPort;
    /**
     * 报文载荷长度。
     */
    private int length;
    /**
     * ACK报文标识。
     */
    private boolean ack;
    /**
     * FIN报文标识，
     */
    private boolean fin;
    /**
     * SYN报文标识。
     */
    private boolean syn;
    /**
     * RST报文标识。
     */
    private boolean rst;
    /**
     * 序列号。
     */
    private long seqNum;
    /**
     * 确认号。
     */
    private long ackNum;
    /**
     * 报文到达时间戳。
     */
    private long arriveTimestamp;
    /**
     * 报文体。
     */
    private String body;

}

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 = "从网络包解析TCP报文失败";
        log.error(message);
        throw new RuntimeException(message, e);
    }
}

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);
}

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;
}

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;
}