偶尔从套接字读取数据而不关闭它

Question

信息

我试图找到一种方法，以设定的时间间隔从传入的套接字流中读取数据块，但忽略其余数据并且不关闭读取之间的连接。我想知道是否有人有一些建议？

我问的原因是我得到了一个网络连接的模数转换器 (ADC)，我想编写一个简单的示波器应用程序。

基本上，一旦我连接到 ADC 并发送一些初始化命令，它需要几分钟才能稳定，此时它开始以字节流的形式输出测量值。

我想每隔几秒读取 1MB 的数据并丢弃其余数据，如果我不丢弃其余数据，ADC 将缓冲 512kB 的读数然后暂停，因此任何后续读取都将是旧数据。如果我关闭读取 ADC 之间的连接，则需要一段时间才能再次发送数据。

问题

我写了一个简单的 Python 脚本作为测试，在这个测试中，我使用了一个连续运行的线程，如果设置了一个标志，它会读取字节到一个未使用的缓冲区，这似乎工作正常。

当我在 Android 上尝试此操作时遇到了问题，因为似乎只有部分数据被丢弃，如果更新间隔太长，ADC 仍会暂停。

我在哪里犯了错误？我的第一个猜测是同步，因为我不确定它是否按预期工作（请参阅 ThreadBucket 类）。我不得不承认我花了很多时间来玩这个，尝试不同的同步排列、缓冲区大小、BufferedInputStream 和 NIO，但没有运气。

对此的任何意见将不胜感激，我不确定在 Java 中使用这样的线程是否是正确的方法。

代码

Reader 类设置线程，连接到 ADC，根据请求读取数据并在其间激活位桶线程（为了清楚起见，我省略了初始化和关闭）。

class Reader {

    private static final int READ_SIZE = 1024 * 1024;

    private String mServer;
    private int mPort;
    private Socket mSocket;
    private InputStream mIn;
    private ThreadBucket mThreadBucket;
    private byte[] mData = new byte[1];
    private final byte[] mBuffer = new byte[READ_SIZE];

    Reader(String server, int port) {
        mServer = server;
        mPort = port;
    }

    void setup() throws IOException {

        mSocket = new Socket(mServer, mPort);
        mIn = mSocket.getInputStream();
        mThreadBucket = new ThreadBucket(mIn);
        mThreadBucket.start();

        // Omitted: Send a few init commands a look at the response

        // Start discarding data
        mThreadBucket.bucket(true);
    }

    private int readRaw(int samples) throws IOException {

        int current = 0;

        // Probably fixed size but may change
        if (mData.length != samples)
            mData = new byte[samples];

        // Stop discarding data
        mThreadBucket.bucket(false);
        // Read in number of samples to mData
        while (current < samples) {
            int len = mIn.read(mBuffer);
            if (current > samples)
                current = samples;
            if (current + len > samples)
                len = samples - current;
            System.arraycopy(mBuffer, 0, mData, current, len);
            current += mBuffer.length;
        }
        // Discard data again until the next read
        mThreadBucket.bucket(true);

        return current;
    }
}

ThreadBucket 类持续运行，如果mBucket 为真，则将数据发送到位桶。

同步是为了阻止其中一个线程读取数据，而另一个线程正在读取数据。

public class ThreadBucket extends Thread {

    private static final int BUFFER_SIZE = 1024;

    private final InputStream mIn;
    private Boolean mBucket = false;
    private boolean mCancel = false;

    public ThreadBucket(final InputStream in) throws IOException {

        mIn = in;
    }

    @Override
    public void run() {

        while (!mCancel && !Thread.currentThread().isInterrupted()) {
            synchronized (this) {
                if (mBucket)
                    try {
                        mIn.skip(BUFFER_SIZE);
                    } catch (final IOException e) {
                        break;
                    }
                }
            }
        }

    public synchronized void bucket(final boolean on) {

        mBucket = on;
    }

    public void cancel() {

        mCancel = true;
    }
}

谢谢。

Answer 1

您需要不断地阅读，周期，尽可能快地编写代码，然后分别管理您对数据的处理方式。不要将两者混为一谈。