java - 在其他线程中使用蓝牙连接的OutputStream会干扰消息的接收

Question

我按照 Android 指南建立了蓝牙连接。

为了分离事物并使它们独立，我决定将 BT 的发送部分放在一个单独的线程中。为了实现这一点，我将 BT-Socket 的“OutStream”传递给一个单独的 Thread 类。我的问题是，一旦我启动这个线程，传入的消息就不再是红色的了。

但我不知道为什么，因为我现在不使用这个线程。它已启动，但未在其中写入任何消息。

这是接收消息的“ConnectedToDevice”类的一部分。我使用一种特殊的方法来检测我的消息是否已完全收到。

public void run() {
    byte[] buffer = new byte[1024];
    int bytes;

    sendPW();
    int len = 0;

    Communication.getInstance().setFrequentSending(OVS_CONNECTION_IN_PROGRESS);
    Communication.getInstance().setSendingMessages(mmOutStream); //Passing the OutStream to the sending class.
    Communication.getInstance().setReceivingMessages(queueReceivingMsg);

    Communication.getInstance().startThreads(); //currently: only start sending thread.

    while (true) {
      try {
        bytes = this.mmInStream.read(buffer, len, buffer.length - len);
        len += bytes;

        if (len >= 3 && buffer[2] != -1) {
          len = 0;
          Log.d(TAG, "run: To Short?");
        } else if (len >= 5) {
          int dataLength = Integer
              .parseInt(String.format("%02X", buffer[3]) + String.format("%02X", buffer[4]), 16);
          if (len == 6 + dataLength) {
            queueReceivingMsg.add(buffer);
            Log.d(TAG, "run: Added to queue");
            len = 0;
          }
          Log.d("BSS", "dataLenght: " + Integer.toString(dataLength) + " len " + len);
        }
      } catch (IOException var5) {
        cancel();
        Communication.getInstance().interruptThreads();
        return;
      }
    }
  }

发送消息类的重要部分

public static BlockingQueue<Obj_SendingMessage> sendingMessages = new LinkedBlockingQueue<>();

@Override
  public void run() {
    while (!isInterrupted()) {
      if (bGotResponse){
        try{
          sendingMessage = sendingMessages.take();
          send(sendingMessage.getsData());
          bGotResponse = false;
          lTime = System.currentTimeMillis();
        } catch (InterruptedException e){
          this.interrupt();
        }
      }

      if((System.currentTimeMillis()%500 == 0) && System.currentTimeMillis() <= lTime+1000){
        if(sendingMessage != null){
          send(sendingMessage.getsData());
        }
      } else {
        bGotResponse =true;
      }
    }
  }

//Where the outStream is used
private void write(int[] buffer) {
    try {
      for (int i : buffer) {
        this.mmOutputStream.write(buffer[i]);
      }
    } catch (IOException var3) {

    }
  }

再次澄清，sendingMessages 始终为空，但消息仍然无法正确接收。

Answer 1

这里提出了从流中读取消息的健壮代码的外观。该代码可以通过以下方式处理部分消息和多条消息:

• 如果消息不完整，则等待更多数据
• 如果有多条消息的数据可用，则处理第一条消息并保存其余数据
• 搜索标记字节 0xff，并在需要丢弃无效数据时保留下一个可能有效消息的数据

在编写这段代码时，我注意到代码中的另一个错误。如果找到消息，则不会复制数据。相反，缓冲区被返回。但是，在处理前一条消息之前或处理前一条消息时，缓冲区以及因此返回的消息可能会被下一条消息覆盖。

这个错误比流数据的不良解码更严重。

private byte[] buffer = new byte[1024];
private int numUnprocessedBytes = 0;

public void run() {
    ...
    while (true) {
        try {
            int numBytesRead = mmInStream.read(buffer, numUnprocessedBytes, buffer.length - numUnprocessedBytes);
            numUnprocessedBytes += numBytesRead;
            processBytes();
        } catch (IOException e) {
            ...
        }
    }
}

private void processBytes() {
    boolean tryAgain;
    do {
        tryAgain = processSingleMessage();
    } while (tryAgain);
}

private boolean processSingleMessage() {
    if (numUnprocessedBytes < 5)
        return false; // insufficient data to decode length

    if (buffer[2] != (byte)0xff)
        // marker byte missing; discard some data
        return discardInvalidData();

    int messageLength = (buffer[3] & 0xff) * 256 + (buffer[4] & 0xff);
    if (messageLength > buffer.length)
        // invalid message length; discard some data
        return discardInvalidData();

    if (messageLength > numUnprocessedBytes)
        return false; // incomplete message; wait for more data

    // complete message received; copy it and add it to the queue
    byte[] message = Arrays.copyOfRange(buffer, 0, messageLength);
    queueReceivingMsg.add(message);

    // move remaining data to the front of buffer
    if (numUnprocessedBytes > messageLength)
        System.arraycopy(buffer, messageLength, buffer, 0, numUnprocessedBytes - messageLength);
    numUnprocessedBytes -= messageLength;

    return numUnprocessedBytes >= 5;
}

private boolean discardInvalidData() {
    // find marker byte after index 2
    int index = indexOfByte(buffer, (byte)0xff, 3, numUnprocessedBytes);
    if (index >= 3) {
        // discard some data and move remaining bytes to the front of buffer
        System.arraycopy(buffer, index - 2, buffer, 0, numUnprocessedBytes - (index - 2));
        numUnprocessedBytes -= index - 2;
    } else {
        // discard all data
        numUnprocessedBytes = 0;
    }
    return numUnprocessedBytes >= 5;
}

static private  int indexOfByte(byte[] array, byte element, int start, int end) {
    for (int i = start; i < end; i++)
        if (array[i] == element)
            return i;

    return -1;
}