在其他线程中使用蓝牙连接的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;
}