5

我是正确的吗？我假设在同一个进程的范围内，有 2 个线程读取/写入命名管道根本不会阻塞读取器/写入器？因此，如果时间错误，可能会丢失一些数据？

并且在多个进程的情况下 - 阅读器将等到一些数据可用，而写入器将被阻止，直到阅读器读取阅读器提供的所有数据？

我打算使用命名管道从外部进程传递几个（数十、数百个）文件，并在我的 Java 应用程序中使用这些文件。编写简单的单元测试以使用一个线程写入管道，另一个线程用于从管道读取，由于缺少数据块而导致零星的测试失败。

我认为这是因为线程和相同的过程，所以我的测试通常不正确。这个假设正确吗？

这是说明这种情况的某种示例：

import java.io.{FileOutputStream, FileInputStream, File}
import java.util.concurrent.Executors
import org.apache.commons.io.IOUtils
import org.junit.runner.RunWith
import org.scalatest.FlatSpec
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class PipeTest extends FlatSpec {

  def md5sum(data: Array[Byte]) = {
    import java.security.MessageDigest
    MessageDigest.getInstance("MD5").digest(data).map("%02x".format(_)).mkString
  }

  "Pipe" should "block here" in {
    val pipe = new File("/tmp/mypipe")
    val srcData = new File("/tmp/random.10m")
    val md5 = "8e0a24d1d47264919f9d47f5223c913e"
    val executor = Executors.newSingleThreadExecutor()
    executor.execute(new Runnable {
      def run() {
        (1 to 10).foreach {
          id =>
            val fis = new FileInputStream(pipe)
            assert(md5 === md5sum(IOUtils.toByteArray(fis)))
            fis.close()
        }
      }
    })
    (1 to 10).foreach {
      id =>
        val is = new FileInputStream(srcData)
        val os = new FileOutputStream(pipe)
        IOUtils.copyLarge(is, os)
        os.flush()
        os.close()
        is.close()
        Thread.sleep(200)
    }
  }

}

没有Thread.sleep(200)测试由于原因未能通过

  • 断管异常
  • 不正确的 MD5 和

有了这个延迟集 - 它工作得很好。我正在使用包含 10 兆字节随机数据的文件。

4

2 回答 2

5

这是您的代码中非常简单的竞争条件：您正在将固定大小的消息写入管道，并假设您可以读回相同的消息。但是，对于任何给定的读取，您不知道管道中有多少可用数据。

如果您在写入前加上写入的字节数，并确保每个读取只读取该字节数，您将看到管道完全按照宣传的方式工作。

如果您遇到多个写入者和/或多个读取者的情况，我建议使用实际的消息队列。实际上，我建议在任何情况下都使用消息队列，因为它解决了消息边界划分的问题；重新发明那个特定的轮子没有什么意义。

于 2013-05-16T16:52:49.990 回答
1

我是正确的吗？我假设在同一个进程的范围内，有 2 个线程读取/写入命名管道根本不会阻塞读取器/写入器？

除非您使用非阻塞 I/O，否则不会。

因此，如果时间错误，可能会丢失一些数据？

除非您使用非阻塞 I/O，否则不会。

于 2013-05-16T21:07:09.273 回答