如果'x'时间已经过去,则创建一个取消InputStream.read()调用的线程

Question

我目前有一个来自Android的BluetoothChat示例的工作I/O流,但遇到了问题.我的应用程序通过蓝牙连接到蓝牙模块,蓝牙模块又向模块物理连接的设备发送信号.

我的程序调用read()输入流,如果有数据发送,程序可以顺利执行,没有任何问题.但是,实现流的方式不能防止中断连接.如果从设备中物理移除模块,或者设备没有发回任何信号,我的代码就会坐下来等待InputStream.read()通话.

我的read()电话看起来像这样:

try {
    Log.i( "1) I/O", "available bits: " + mmInStream.available() );
    bytes = mmInStream.read(buffer, 0, length);
    Log.i( "2) I/O", "available bits: " + mmInStream.available() );
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                        .sendToTarget();
} catch (Exception e) {
    Log.i(TAG,  "Catch Statement" );
    Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
    Bundle bundle = new Bundle();
    bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
    msg.setData(bundle);
    mHandler.sendMessage(msg);
    Log.e(TAG, "disconnected a", e);
    connectionLost();

    // Start the service over to restart listening mode
    BluetoothService.this.start();
    //break;
}

当我的程序正常运行时,块Log中的两个调用都try返回0for的值mmInStream.available().当输入流被中断时,初始Log调用返回a 0,而第二个调用永远不会被调用.然后我的程序在catch每次到达块之前都会崩溃.

我一直在寻找几天来解决这个问题,并找到了许多解决方案,但它们要么没有用,要么我不理解它们.

1)使用扫描仪输入InputStream如下所示.这没有提供帮助,也在阅读时超时.

Scanner scan = new Scanner(new InputStreamReader(mmInStream));
scan.useDelimiter( "[\\r\\n]+" );
String readIn;

try {
    readIn = scan.next();
    scan = null;
    tempB = readIn.getBytes( Charset.forName( "US-ASCII" ) );
    append = "\r\n".getBytes( Charset.forName( "US-ASCII" ) );
    for( int i = 0; i < length; i++ ) {
        if( i == length - 1 ) {
            buffer[i] = append[1];
        } else if ( i == length - 2 ) {
            buffer[i] = append[0];
        } else {
            buffer[i] = tempB[i];
        }
    }
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                        .sendToTarget();
} catch (Exception e) {
    Log.i(TAG,  "Catch Statement" );
                Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
                Bundle bundle = new Bundle();
                bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
                msg.setData(bundle);
                mHandler.sendMessage(msg);
                Log.e(TAG, "disconnected a", e);
                connectionLost();

                // Start the service over to restart listening mode
                BluetoothService.this.start();
                //break;
            }

2)我已经尝试运行一个线程,它会read在X时间后取消调用,但它无法正常工作:

public void run(int length) throws IOException {
    buffer = new byte[1024];
    length1 = length;
    Thread myThread = new Thread(new Runnable() {
        public void run() {
            try {
                bytes = mmInStream.read( buffer, 0, length1 );
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    });

    synchronized (myThread) {
        myThread.start();
        try {
            myThread.wait(500);
            if(myThread.isAlive()) {
                mmInStream.close();
                Log.i( "InStream", "Timeout exceeded!");
            }
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }
   try {
        myThread.run();
        mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                    .sendToTarget();
   } catch (IOException e) {
            Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
            Bundle bundle = new Bundle();
            bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
            msg.setData(bundle);
            mHandler.sendMessage(msg);
            connectionLost();
            BluetoothService.this.start();
   }

在这两个选项不起作用之后,我一直试图调查Java NIO或者AsyncTask,但是所有这些似乎都是为了识别I/O超时而添加的东西.我也看到有些Sockets支持使用超时功能.setSoTimeout(),但是这是一个BluetoothSocket我发现他们不支持此功能.

由于没有I/O类支持read()将超时长度作为参数或超时的方法,因此在我看来,添加一个Thread将是最简单的实现.这是错的吗？关于我对上述方法做错了什么的信息,或者如何合并Java NIO/ AsyncTask将会非常感激.

编辑:

这是我尝试的新线程代码,我目前正在将其更改为给定答案显示并尝试该操作.如果以后它不起作用我会张贴.

Thread myThread = new Thread(new Runnable() {
            public void run() {
                try {
                    bytes = mmInStream.read( buffer, 0, length1 );
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });

        synchronized (myThread) {
            try {
                myThread.wait(6000);
                Log.i( "InStream", "After wait" );
                if(myThread.isAlive()) {
                    Log.i( "InStream", "Timeout exceeded2!");
                    myThread.interrupt();
                    Log.i( "InStream", "Timeout exceeded!");
                } else {
                    myThread.interrupt();
                }
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                Log.i( "InStream", "Exception Caught" );
                e.printStackTrace();
            }

        }

编辑2:

我试过Dheerej下面给出的答案.我得到IllegalMonitorStateException了wait()函数调用.我尝试了,因为它在答案中显示,然后也尝试myThread.wait()而不是Thread.currentThread.wait().我假设抛出此异常,因为这是myThread在另一个线程中创建并运行的对象.无论如何,下面的代码几乎与Dheerej's答案相同.

        int length1 = length;
            Thread myThread = new Thread(new Runnable() {
                public void run() {
                    buffer = new byte[1024];
                    try {
                        bytes = mmInStream.read(buffer, 0, length1);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                                .sendToTarget();
                }
            });

            myThread.start();
            try {
                //Thread.currentThread().wait(500);
                myThread.wait( 1000 );              // Line 533
            } catch (InterruptedException e) {
                e.printStackTrace();
                //Log.i(TAG,  "Catch Statement" );
                Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
                Bundle bundle = new Bundle();
                bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
                msg.setData(bundle);
                mHandler.sendMessage(msg);
                Log.e(TAG, "disconnected a", e);
                connectionLost();

                // Start the service over to restart listening mode
                BluetoothService.this.start();
            }

            if (myThread.isAlive()) {
                mmInStream.close(); // Alternatively try: myThread.interrupt()
            }

这是生成的LogCat.错误说它从第533行开始,这是wait()上面的调用:

12-28 17:44:18.765: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: return 1
12-28 17:44:18.765: D/BLZ20_WRAPPER(3242): blz20_wrp_write: wrote 3 bytes out of 3 on fd 62
12-28 17:44:18.769: W/NATIVE CODE(3242): -4) baud9600=1, goodbaud=1
12-28 17:44:18.769: D/AndroidRuntime(3242): Shutting down VM
12-28 17:44:18.769: W/dalvikvm(3242): threadid=1: thread exiting with uncaught exception (group=0x40015578)
12-28 17:44:18.773: E/AndroidRuntime(3242): FATAL EXCEPTION: main
12-28 17:44:18.773: E/AndroidRuntime(3242): java.lang.IllegalMonitorStateException: object not locked by thread before wait()
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.Object.wait(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.Object.wait(Object.java:395)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService$ConnectedThread.run(BluetoothService.java:533)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService.read(BluetoothService.java:326)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService.changeitJava(BluetoothService.java:669)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.RelayAPIModel$NativeCalls.changeItJavaWrapper(RelayAPIModel.java:490)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.RelayAPIModel$NativeCalls.InitRelayJava(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.MainMenu$1.handleMessage(MainMenu.java:547)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.os.Handler.dispatchMessage(Handler.java:99)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.os.Looper.loop(Looper.java:130)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.app.ActivityThread.main(ActivityThread.java:3687)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.reflect.Method.invokeNative(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.reflect.Method.invoke(Method.java:507)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:842)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:600)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at dalvik.system.NativeStart.main(Native Method)
12-28 17:44:18.781: D/BLZ20_ASOCKWRP(3242): asocket_read
12-28 17:44:18.781: I/BLZ20_WRAPPER(3242): blz20_wrp_poll: nfds 2, timeout -1 ms
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: transp poll : (fd 62) returned r_ev [POLLIN ] (0x1)
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: return 1
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_read: read 5 bytes out of 5 on fd 62

Answer 1

先试试这个:

try {
    int available = 0;

    while (true)
    {
        int available = mmInStream.available();
        if (available > 0) { break; }
        Thread.sleep(1);
        // here you can optionally check elapsed time, and time out
    }

    Log.i( "1) I/O", "available bits: " + available );
    bytes = mmInStream.read(buffer, 0, length);
    Log.i( "2) I/O", "available bits: " + mmInStream.available() );
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer).sendToTarget();
} catch (Exception e) {
    ...
}

在您available()之前调用的原始代码中,read()通常没有数据等待读取.然后你调用read(),阻塞并等待数据,然后读取所有数据.然后available()再次调用,再一次没有数据,因为它已全部被读取:)更好:睡眠直到available()返回非零,然后读取.然而,这可能无法工作,因为available()是总是允许返回0(即使数据实际上是可用).

如果上述方法不起作用,请尝试此问题的技巧:是否可以从具有超时的InputStream读取？

Callable<Integer> readTask = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        return mmInStream.read(buffer, 0, length);
    }
}

try {
    Future<Integer> future = executor.submit(readTask);
    bytes = future.get(100, TimeUnit.MILLISECONDS);
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer).sendToTarget();
} catch (TimeoutException e) {
    // deal with timeout in the read call
} catch (Exception e) {
    ...
}

最后,BluetoothSocket文档说你可以从任何线程关闭套接字,并立即生效.所以你可以简单地有一个看门狗线程,如果读取调用没有成功,则调用close()套接字会导致阻塞read()返回错误.这就是Dheeraj上面提到的,但你只需要close()在另一个线程被卡住时调用(由于网络错误/连接丢失/等):否则只需偶尔检查一下它的进度但是只要你没有关闭阅读时间不长.

这当然看起来缺乏超时(和()从外部中断阻塞读是不可能的),已经在Java中的一个正在进行的主要痛点长的时间.

也可以看看:

是否可以从具有超时的InputStream读取？(用途Callable/ Future)

我可以为InputStream的read()函数设置超时吗？(用途Socket.setSoTimeout())

如何杀死BufferedInputStream .read()调用(使用InterruptibleChannel)

如何在Java中阻止线程在阻塞读取操作中等待？