Rx读取文件并将数据发送给12个观察者

Question

解释最后一段代码：

在我使用下面的代码来读取和发送这个日期之前，白色处理程序。

代码读取二进制文件，分配8* 12*1002字节，为信号数据(包含12 * 1002双精度值)

问题：

如何使用rx_java2读取此文件和流循环，同步每1002双同步到12个观察者，延迟为每2ms发布双精度值

class readFile {


public fun getBuffer(sigId: String): ArrayList<DoubleArray> {

    
  val address="/data/data/com.example.amin.ecgs/Signals/Sig$sigId.bin"
    lateinit var buf: DoubleBuffer
    try {

        val rFile = RandomAccessFile(address, "rw")
        val inChannel = rFile.channel
        val buf_in = ByteBuffer.allocate(8 * 12 * 1002)
        buf_in.clear()
        inChannel.read(buf_in, 0)
        buf_in.rewind()
        buf = buf_in.asDoubleBuffer()

        inChannel.close()

    } catch (e: IOException) {
        e.printStackTrace()

    }

    return generateData(buf)

}

private fun generateData(buf: DoubleBuffer): ArrayList<DoubleArray> {

    val arrayList = ArrayList<DoubleArray>(12)

    for (n in 0..11) {
        val yb = DoubleArray(1002)
        buf.get(yb, n * 1002, (n + 1) * 1002)

        arrayList.add(yb)

    }
    buf.clear()
    return arrayList
}


}

Answer 1

您可以尝试使用进行concat操作，方法是创建每个Observable或从Observable.fromIterator()创建一些内容，如下所示。请注意，当您这样做时以及当您必须处理下游错误时，错误处理会变得棘手。使用Flowable会更好，因为如果您希望在其中一个数组失败的情况下继续使用下一个数组，则可以使用onErrorResumeNext定义的BackPressureStrategy。

public class ReadBinaryDate extends Thread {

    final ArrayList<Observable<double[]>> doubleObservables = new ArrayList<>();

    private Observable<double[]> generateObservable (final double [] array1002) {
        return Observable.interval(2, TimeUnit.MILLISECONDS)
            .flatMap(new Function<Long, ObservableSource<double[]>>() {
                @Override
                public ObservableSource<double[]> apply(Long aLong) throws Exception {
                    return Observable.just(array1002);
                }
            });
    }

    private void generateData() {

        arrayList.clear();
        for (int n = 0; n < 12; n++) {
            double[] yb = new double[1002];
            buf.get(yb, n*1002, (n+1)*1002);
            doubleObservables.add(generateObservable(yb));
        }
    }

    private void subscribeStream() {
        Observable.fromIterable(doubleObservables)
            .subscribe(new Observer<Observable<double[]>>() {
                @Override
                public void onSubscribe(Disposable d) { }

                @Override
                public void onNext(Observable<double[]> observable) {
                    // do on each array
                }

                @Override
                public void onError(Throwable e) {
                    // handle error
                }

                @Override
                public void onComplete() {
                    // do on complete
                }
            });
    }
}