PersistentBuffer 是一系列 Akka Streams 流组建的第一个。它像 Akka Streams缓冲区一样工作，不同的是，缓存的内容存储于一系列内存映射的文件中，由 PersistentBuffer构造提供。这可以让缓冲区大小实际无上限，不使用JVM堆来存储，在同时处理百万级/每秒时性能优异。
依赖
以下依赖在Persistent Buffer工作时需要:
"org.squbs" %% "squbs-pattern" % squbsVersion,
"net.openhft" % "chronicle-queue" % "4.5.13"
例子
以下的例子显示了 PersistentBuffer在流中的用法：
implicit val serializer = QueueSerializer[ByteString]()
val source = Source(1 to 1000000).map { n => ByteString(s"Hello $n") }
val buffer = new PersistentBuffer[ByteString](new File("/tmp/myqueue"))
val counter = Flow[Any].map( _ => 1L).reduce(_ + _).toMat(Sink.head)(Keep.right)
val countFuture = source.via(buffer.async).runWith(counter)
此版本在GraphDSL显示相同
implicit val serializer = QueueSerializer[ByteString]()
val source = Source(1 to 1000000).map { n => ByteString(s"Hello $n") }
val buffer = new PersistentBuffer[ByteString](new File("/tmp/myqueue"))
val counter = Flow[Any].map( _ => 1L).reduce(_ + _).toMat(Sink.head)(Keep.right)
val streamGraph = RunnableGraph.fromGraph(GraphDSL.create(counter) { implicit builder =>
sink =>
import GraphDSL.Implicits._
source ~> buffer.async ~> sink
ClosedShape
})
val countFuture = streamGraph.run()
背压（Back-Pressure）
PersistentBuffer不背压上游流量。它将获取给它的所有的流元素，并且通过增长或旋转队列文件的数量来增长存储。它没有任何方法确定缓冲区的限制或存储大小。下游流量背压按照每个Akka Streams和Reactive Streams的要求进行。
如果PersistentBuffer stage被下游流量混淆， PersistentBuffer不会缓存并且它实际上会背压。为了保证PersistentBuffer确实运行在它自己的空间，在这之后加入一个async 边界。
失败 & 恢复
由于它的持久特性， PersistentBuffer可以从突然的流关闭，故障，JVM故障甚至潜在的系统故障中恢复。在同一个目录通过 PersistentBuffer重启流将启动发出存贮在缓存中的元素，在新的元素加入进来之前不会消费。在先前的流故障或关闭时缓存中正在消费的元素（并未消费完成）将会丢失。
因为缓存通过本地存储、心轴或SSD，因此这个缓存的性能和耐久性同样取决于存储的耐久性。所以，理解和推断缓存的耐久性非常重要，与那些数据库和其他热离线持久存储不是同一个级别，以换取更高的性能。
Akka Streams stage批处理请求并在内部缓存记录。 PersistentBuffer保证回复和记录的持久化到达onPush， Akka Stream stage内部缓存的记录未到达onPush 将会在故障中丢失。
提交保证（Commit Guarantee）
在一个不可预知的故障情况中，从 PersistentBuffer stage发出的元素却未抵达sink将会丢失。有些情况下，它可能需要避免此类数据丢失。在 sink之前使用commit stage对这类情况有帮助。加入 commit stage，使用 PersistentBufferAtLeastOnce 。请参考下面commit stage 用法的例子：
implicit val serializer = QueueSerializer[ByteString]()
val source = Source(1 to 1000000).map { n => ByteString(s"Hello $n") }
val tempPath = new File("/tmp/myqueue")
val config = ConfigFactory.parseMap {
Map(
"persist-dir" -> s"${tempPath.getAbsolutePath}"
)
}
val buffer = new PersistentBufferAtLeastOnce[ByteString](config)
val commit = buffer.commit[ByteString]
val flowSink = // do some transformation or a sink flow with expected failure
val counter = Flow[Any].map( _ => 1L).reduce(_ + _).toMat(Sink.head)(Keep.right)
val streamGraph = RunnableGraph.fromGraph(GraphDSL.create(counter) { implicit builder =>
sink =>
import GraphDSL.Implicits._
// ensures that records are reprocessed when something fails at tranform flow
source ~> buffer ~> flowSink ~> commit ~> sink
ClosedShape
})
val countFuture = streamGraph.run()
请注意，commit 无法防止在 sink（或者其他commit之后的stage）内部缓存中的丢失。
提交订单（Commit Order）
commit stage应该正常按照顺序接收元素。然而，流中的一个潜在的bug可能引起一个元素丢弃或不按顺序抵达 commit stage。默认的commit-order-policy设置为 lenient，使流继续运行在这个场景中。你可以设置为 strict，以便抛出CommitOrderException 异常，并让Supervision.Decider确定要执行的操作。
空间管理
一个典型的持久队列目录查看如下：
$ ls -l
-rw-r--r-- 1 squbs_user 110054053 83886080 May 17 20:00 20160518.cq4
-rw-r--r-- 1 squbs_user 110054053 8192 May 17 20:00 tailer.idx
当所有的读者成功处理读取queue，队列文件自动删除。
配置
队列通过传递一个保存了所有默认配置的持久化目录的地址创建。所有的例子可以在上面看到。或者，它可以通过传递在构建时一个 Config对象创建。 Config 对象是一个标准的HOCON 配置。下面的例子展示了使用Config构建PersistentBuffer：
val configText =
"""
| persist-dir = /tmp/myQueue
| roll-cycle = xlarge_daily
| wire-type = compressed_binary
| block-size = 80m
""".stripMargin
val config = ConfigFactory.parseString(configText)
//使用Config构建缓存
val buffer = new PersistentBuffer[ByteString](config)
下面的配置属性用于 PersistentBuffer
persist-dir = /tmp/myQueue # Required
roll-cycle = daily # Optional, defaults to daily
wire-type = binary # Optional, defaults to binary
block-size = 80m # Optional, defaults to 64m
index-spacing = 16k # Optional, defaults to roll-cycle's spacing
index-count = 16 # Optional, defaults to roll-cycle's count
commit-order-policy = lenient # Optional, default to lenient
Roll-cycle可以用大写或小写指定。roll-cycle支持以下值：
| Roll Cycle | 容量（Capacity） |
|---|---|
| MINUTELY | 64 million entries per minute |
| HOURLY | 256 million entries per hour |
| SMALL_DAILY | 512 million entries per day |
| DAILY | 4 billion entries per day |
| LARGE_DAILY | 32 billion entries per day |
| XLARGE_DAILY | 2 trillion entries per day |
| HUGE_DAILY | 256 trillion entries per day |
Wire-type可以通过大写或者小写指定。wire-type支持以下值：
- TEXT
- BINARY
- FIELDLESS_BINARY
- COMPRESSED_BINARY
- JSON
- RAW
- CSV
内存大小诸如 block-size 和index-spacing 依据memory size format defined in the HOCON specification指定。
序列化（Serialization）
QueueSerializer[T] 需要被隐式的提供给PersistentBuffer[T]，如上面的例子所示：
implicit val serializer = QueueSerializer[ByteString]()
QueueSerializer[T]() 为你的目标类型调用生产一个序列化器（Serializer）。它基于基础设施的序列化和反序列化。
实现Serializer
控制队列中细粒度的持久化格式，你可能需要实现你自己的序列化器（serializer）如下：
case class Person(name: String, age: Int)
class PersonSerializer extends QueueSerializer[Person] {
override def readElement(wire: WireIn): Option[Person] = {
for {
name <- Option(wire.read().`object`(classOf[String]))
age <- Option(wire.read().int32)
} yield { Person(name, age) }
}
override def writeElement(element: Person, wire: WireOut): Unit = {
wire.write().`object`(classOf[String], element.name)
wire.write().int32(element.age)
}
}
使用这个序列化器（serializer），只需要在构建PersistentBuffer之前声明它为隐式的，如下：
implicit val serializer = new PersonSerializer()
val buffer = new PersistentBuffer[Person](new File("/tmp/myqueue")
广播缓存（Broadcast Buffer）
BroadcastBuffer是持久化缓存的一个变种。这个工作与PersistentBuffer相似，流元素广播至多个输出端口。因此它是缓存和广播stage的组合。这个配置采用一个名为output-ports的附加参数，用于指定输出端口的数量。
当流元素从每个输出端口发出（以独立的速度，取决于下游的速度要求）时，特别需要广播缓存。
val configText =
"""
| persist-dir = /tmp/myQueue
| roll-cycle = xlarge_daily
| wire-type = compressed_binary
| block-size = 80m
| output-ports = 3
""".stripMargin
val config = ConfigFactory.parseString(configText)
// Construct the buffer using a Config.
val bcBuffer = new BroadcastBuffer[ByteString](config)
例子
implicit val serializer = QueueSerializer[ByteString]()
val in = Source(1 to 100000)
val flowCounter = Flow[Any].map(_ => 1L).reduce(_ + _).toMat(Sink.head)(Keep.right)
val streamGraph = RunnableGraph.fromGraph(GraphDSL.create(flowCounter) { implicit builder =>
sink =>
import GraphDSL.Implicits._
val buffer = new BroadcastBufferAtLeastOnce[ByteString](config)
val commit = buffer.commit[ByteString]
val bcBuffer = builder.add(buffer.async)
val mr = builder.add(merge)
in ~> transform ~> bcBuffer ~> commit ~> mr ~> sink
bcBuffer ~> commit ~> mr
bcBuffer ~> commit ~> mr
ClosedShape
})
val countFuture = streamGraph.run()
积分（Credits）
PersistentBuffer 利用Chronicle-Queue 4.x作为高性能内存映射队列持久化。
