Java 多播发送方 + 接收方

Question

我的问题很容易解释。我有一个多播发送方和一个多播接收方。我试图发送一个 8MB 的大文件。它应该分成 1024 字节包 + 4 字节 header 。发送正常，但接收方有时会在 2000 或 3000 取消接收位置 5000。我不知道为什么它没有收到所有元素。

发件人:

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;

public class Sender extends Thread{

protected static MulticastSocket socket = null;
public int QuoteCount = 0;
public int Time_Interval = 6000;
public static FileInputStream in;
final static int split_size = 1028 ;

public static void main(String[] args) throws IOException{
    // args 0 => path
    // args1=> ip multicast
    // args2 => networkinterface ip
    // args3 => port 
    // args4 => ttl
    //socket = new MulticastSocket(444);
    InetAddress ip_address_group = InetAddress.getByName(args[1]);
    System.out.println("Wait 4 clients to connect...");
    File file=new File(args[0]);

    InetSocketAddress address = new InetSocketAddress(args[1],Integer.parseInt(args[3]));
    MulticastSocket socket = new MulticastSocket(new InetSocketAddress(args[2], Integer.parseInt(args[3])));
    socket.connect(address);
    socket.setTimeToLive(Integer.parseInt(args[4]));

    MD5 md5 = new MD5();
    try {
        System.out.println("MD5 vom File: "+md5.getFileMD5(args[0]));
    } catch (Exception e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }

    // Filesize package
    int filesize = (int)file.length();

    double anzpak=(filesize/1028); // anzahl pakete
    double diagrammgrenze=(80/anzpak); // wieviele stricherl setzt ein paket
    double strikecount=0;
    System.out.println("DIAGR: "+diagrammgrenze +" Anzpak: "+anzpak);

    byte[] firstpack=new byte[4];
    int2bytearr(filesize,firstpack);
    DatagramPacket firsttosend=new DatagramPacket(firstpack,firstpack.length,ip_address_group, Integer.parseInt(args[3]));
    socket.send(firsttosend);

    // Rest of packages
    in = new FileInputStream(file);

    byte[] data = new byte[split_size];
    int numbytes = 0;
    int seqnr = 0;
    int sentbytes=0;

    try {
        while((numbytes = in.read(data)) != -1){
            // Generate 4 byte long seqnr:
            seqnr++;
            strikecount+=diagrammgrenze;
            if(strikecount>=1){
                for(int i=0;i<(int)strikecount;i++){
                    System.out.print("|");
                    strikecount--;
                }
            }

            byte[] dataseq = new byte[4];
            int2bytearr(seqnr,dataseq);
            sentbytes+=numbytes;

            // DATA PLUS SEQNR Generation 
            byte[] seqplusdata = new byte[dataseq.length + data.length];
            System.arraycopy(dataseq, 0, seqplusdata, 0, dataseq.length);
            System.arraycopy(data, 0, seqplusdata, dataseq.length, data.length);

            // Data Plus Seqnr Sending
            DatagramPacket tosend=new DatagramPacket(seqplusdata,seqplusdata.length);
            socket.send(tosend);
        }

    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    System.out.println("\nTosend filesize: "+filesize);
    System.out.println("Sent bytes: "+sentbytes);
    //in.close();
    socket.close();
}

public static void int2bytearr(int number,byte[] data){
    for (int i = 0; i < 4; ++i) {
        int shift = i << 3; // i * 8
        data[3-i] = (byte)((number & (0xff << shift)) >>> shift);
    }
}

}

接收者:

import java.io.IOException;
import java.net.MulticastSocket;
import java.net.DatagramPacket;
import java.net.InetAddress;

public class Empfaenger extends Thread{

public static void main(String[] args) throws IOException{
    Empfaenger empfaenger = new Empfaenger();
    empfaenger.start();
}

@SuppressWarnings("resource")
public void run(){
    try{

        //Create socket
        MulticastSocket socket = new MulticastSocket(12345);

        //Connect to server (must be multicast)
        InetAddress IP_Adress = InetAddress.getByName("228.5.6.7");
        socket.joinGroup(IP_Adress);

        DatagramPacket packet;
        int pcount=0;

        // firstpack for getting filesize package
        byte[] firstpack = new byte[4];
        DatagramPacket firstpacket=new DatagramPacket(firstpack,firstpack.length);
        socket.receive(firstpacket);
        int filesize=makeintfrombyte(firstpack);
        System.out.println("Empfaenger filesize: " + filesize);

        double anzpak=(filesize/1028); // anzahl pakete
        double diagrammgrenze=(80/anzpak); // wieviele stricherl setzt ein paket
        double strikecount=0;

        for(int i=0;i<anzpak;i++){
            System.out.println(pcount + "< "+anzpak);

            strikecount+=diagrammgrenze;
            if(strikecount>=1){
                while(strikecount>=1){
                    System.out.print("|");
                    strikecount--;
                }
            }

            byte[] buf = new byte[1028];
            packet = new DatagramPacket(buf, buf.length);
            socket.receive(packet);
            pcount++;
            //System.out.println("SeqNr. in Bytes: "+buf[0]+"|"+buf[1]+"|" +buf[2]+"|" +buf[3]+"|" + pcount);
        }
        //socket.leaveGroup(IP_Adress);
        //socket.close();

    }catch ( IOException X) {System.out.println(X);}
}

public int makeintfrombyte(byte[] b){
    return b[0] << 24 | (b[1] & 0xff) << 16 | (b[2] & 0xff) << 8 | (b[3] & 0xff);
}

}

这里是接收方和一个发送方的示例输出] http://i.stack.imgur.com/UblCa.jpg

编辑:如果我们让发件人 sleep (长)1.0 它有效，但这不是让发件人 sleep 的意思:(

Answer 1

数据报不保证交付，因此它不是传输文件的理想协议(protocol)。您可能需要考虑使用可靠的多播协议(protocol)。

最后，如果您的发送速率高于接收速率，您最终会丢弃数据包。事实上，即使速率非常相似，接收端(例如调度程序、垃圾收集器等)的任何临时减速都足以开始下降。通过调用 setReceiveBufferSize 增加底层网络缓冲区大小会给你一些利润并大大改善行为。默认值通常为 128K，至少在 UX 系统上是这样；尝试将其设置为几 MB。