큐 에서 데이터를 읽기 전에 데이터가 있는지 확인했는데도 에러가 발생합니다.

Question

usb 시리얼통신으로 데이터를 받아서 큐에 저장하고 동시에 스레드로 큐에서 데이터를
읽어 들여 처리하는 소스 입니다.
자꾸 스레드가 죽는데요.  try catch 해보니 에러 원인으로
java.lang.IllegalStateException: mavlinkpacketQueue.poll() must not be null  라고 뜨네요.
mavlinkpacketQueue.poll() 사용할때 즉 큐에서 데이터를 꺼네고 삭제하기전에 데이터가
있는지 확인했는데요.  왜 이러한 에러가 발생하는지 모르겠습니다.
고수님들의 지혜가 필요합니다.  감사합니다.

open class BaseSprayActivity : BaseActivity(), SerialInputOutputManager.Listener {

//usb 시리얼 통신 라이브러리를 참조하고 SerialInputOutputManager리스너를 장착합니다.

//SerialInputOutputManager 자체는 스레드입니다.  시리얼 통신으로 데이터를 받으면

//onNewData () 를 자동으로 호출합니다.

 

private var mavlinkpacketQueue: Queue<Byte> = LinkedList()  // 큐를 선언합니다.

 

override fun onNewData(data: ByteArray) {  // 데이터를 수신하면 자동 호출됩니다.

        mainLooper!!.post {

            receive(data)

        }

 }

 

    private fun receive(data: ByteArray) {

        val bufflen = data.size;

        for(i in 0..bufflen-1) {

             var ldata = data.get(i)

             mavlinkpacketQueue.add(ldata) // 큐에 수신된데이터를 입력합니다.

        }

   }

 

inner class readPaketThread: Thread() {

        override fun run() {

            var payload_len : Byte = 0

            var pack_size: Int = 0

            while(isRunning) {

                if(mavlinkpacketQueue.size < 1) continue

                if(mavlinkpacketQueue.peek() == 0xFE.toByte())

                {

                    pack_size = mavlinkpacketQueue.size

                    if(mavlinkpacketQueue.size < 2) continue

                    payload_len = mavlinkpacketQueue.elementAt(1)

                    if(payload_len < 0 || payload_len > 255) { // payload 길이가 잘못 되었다.

                            if(mavlinkpacketQueue.size > 0) {

                                mavlinkpacketQueue.poll()  // FE 삭제...패킷을 버린다.

                                Log.d("delete_poll", "deleate...........................")

                                continue

                            }

                        }

 

                    if(mavlinkpacketQueue.size < payload_len+8)

                        continue

 

                    var data = ByteArray(payload_len.toInt()+8)

 

                        for(j in 0..payload_len+8-1) {

                            if(mavlinkpacketQueue.size > 0) {

                                data[j] = mavlinkpacketQueue.poll()  // 큐 에 저장된 데이터를 읽고 지운다

                            }

                        }

 

                    if(mavlinkpacketQueue.size >= 1000) {  // 큐가 1킬로바이트되면 비운다.

                        mavlinkpacketQueue.clear()

                    }

 

                } else {

                        if(mavlinkpacketQueue.size > 0) {

                            mavlinkpacketQueue.poll()  // 큐에 저장된 데이터를 삭제

                        }

                }

 

            }

 

            super.run()

        }

    }

}

Answer 1

저라면 먼저 데이터를 추가하는 부분과 읽어서 처리하는 부분이 데이터를 공유하지 않도록 변경할 것 같습니다.

?

class SerialDataReceiver: SerialInputOutputManager.Listener {     interface Listener {        fun onReceive(data: ByteArray)     }         private val listeners = hashSetOf<SerialDataReceiver.Listener>()     fun registerListener(listener: SerialInputOut.Listener) {          listeners.add(listener)     }       fun unregisterListener(listener: SerialDataReceiver.Listener) {           val iterator = listeners.iterator()           while (iterator.hasNext()) {               val nextListener = iterator.next()               if (nextListener == listener) {                    iterator.remove()               }           }     }      override fun onNewData(data: ByteArray) {         for (listener in listeners) {             listener.onReceive(data)         }     } }         open class BaseSprayActivity : BaseActivity(), SerialDataReceiver.Listener {      private val serialDataReceiver = SerialDataReceiver()        @Override    fun onResume() {       super.onResume()       serialDataReceiver.registerListener(this)    }           @Override    fun onStop(...) {       super.onStop(...)       serialDataReceiver.unregisterListener(this)    }        override fun onReceive(data: ByteArray) {          // 필요한 처리     }   }

 

만약 위처럼 변수를 공유하지 않고 콜백 내에서 처리하기 곤란하다면, thread 에 안전한 ArrayBlockingQueue나 LinkedBlockginQueue 등과 같은 클래스를 사용하시는 게 나을 듯 합니다.

Comments

네 감사합니다.   나열해주신 해결책들을 적용해 보도록 하겠습니다.

Answer 2

Java 문서에 보시면 LinkedList는 

Note that this implementation is not synchronized. If multiple threads access a linked list concurrently, and at least one of the threads modifies the list structurally, it must be synchronized externally. (A structural modification is any operation that adds or deletes one or more elements; merely setting the value of an element is not a structural modification.) This is typically accomplished by synchronizing on some object that naturally encapsulates the list. If no such object exists, the list should be "wrapped" using the Collections.synchronizedList method. This is best done at creation time, to prevent accidental unsynchronized access to the list:

라고 나와 있습니다. 빨간색 부분을 잘 확인해 보세요. 여러 쓰레드에서 동시에 접근할 경우는  sychronoised 처리를 하라고 되어 있습니다. thread에 안전한 데이터 타입을 찾으셔서 적절한 걸 사용하셔도 될 것 같구요.

Comments

thread는 위 readPaketThread 하나가 전부구요.  onNewData () 가 수시로 콜백되어mavlinkpacketQueue 큐에 데이터를 삽입하게 됩니다.  이런경우 어떻게 동기화 시키죠?  이러한 경우에도 synchronized 를 사용할 수 있나요?

---

큐에 데이터를 집어 넣는 쓰레드와 읽어서 처리하는 쓰레드가 달라 보이는데요.

---

네 맞습니다....둘은 서로 다른 스레드 입니다.