我正在使用来自提供实时流的数据供应商的 Java API。我想使用 Akka 流来处理这个流。
Java API 有一个 pub sub 设计，大致是这样工作的：
Subscription sub = createSubscription();
sub.addListener(new Listener() {
public void eventsReceived(List events) {
for (Event e : events)
buffer.enqueue(e)
}
});
我尝试将此订阅的创建和随附的缓冲区嵌入自定义图形阶段，但没有取得多大成功。谁能指导我使用 Akka 与此 API 交互的最佳方式？ Akka Streams 是这里最好的工具吗？
【问题讨论】：
标签： scala akka akka-stream
要提供Source，您不一定需要使用自定义图形阶段。 Source.queue 将具体化为一个缓冲队列，您可以向其中添加元素，然后这些元素将通过流传播。
有一些棘手的事情需要注意。首先是实现Source.queue 有一些微妙之处，因此您可以设置订阅。像这样的：
def bufferSize: Int = ???
Source.fromMaterializer { (mat, att) =>
val (queue, source) = Source.queue[Event](bufferSize).preMaterialize()(mat)
val subscription = createSubscription()
subscription.addListener(
new Listener() {
def eventsReceived(events: java.util.List[Event]): Unit = {
import scala.collection.JavaConverters.iterableAsScalaIterable
import akka.stream.QueueOfferResult._
iterableAsScalaIterable(events).foreach { event =>
queue.offer(event) match {
case Enqueued => () // do nothing
case Dropped => ??? // handle a dropped pubsub element, might well do nothing
case QueueClosed => ??? // presumably cancel the subscription...
}
}
}
}
)
source.withAttributes(att)
}
Source.fromMaterializer 用于在每次物化时访问物化器（这就是将流定义编译成演员的东西）。当我们物化时，我们使用物化器来preMaterialize队列源，所以我们可以访问队列。我们的订阅将传入元素添加到队列中。
如果消费者跟不上，这个 pubsub 的 API 似乎不支持背压。如果缓冲区已满，队列将丢弃它所传递的元素：在这种情况下，您可能什么都不想做，但我在比赛中已经指出，您应该在这里做出明确的决定。
丢弃最新的元素是这个队列的同步行为（还有其他可用的队列实现，但它们会以异步方式进行通信，这对突发的内存消耗非常不利）。如果您更喜欢其他东西，在队列中设置一个非常小的缓冲区并将“整体”Source（Source.fromMaterializer 返回的那个）附加到表示永久需求的阶段可能是有意义的。例如，buffer(downstreamBufferSize, OverflowStrategy.dropHead) 将丢弃尚未处理的最旧事件。或者，可以以某种有意义的方式组合您的 Events，在这种情况下，如果下游无法快速处理它们，conflate 阶段将自动组合传入的 Events。
【讨论】：
很好的答案！我确实建立了类似的东西。还有一些 kamon 指标来监控队列大小（不包括在内）。
class AsyncSubscriber(projectId: String, subscriptionId: String, metricsRegistry: CustomMetricsRegistry, pullParallelism: Int)(implicit val ec: Executor) {
private val logger = LoggerFactory.getLogger(getClass)
def bufferSize: Int = 1000
def source(): Source[(PubsubMessage, AckReplyConsumer), Future[NotUsed]] = {
Source.fromMaterializer { (mat, attr) =>
val (queue, source) = Source.queue[(PubsubMessage, AckReplyConsumer)](bufferSize).preMaterialize()(mat)
val receiver: MessageReceiver = {
(message: PubsubMessage, consumer: AckReplyConsumer) => {
metricsRegistry.inputEventQueueSize.update(queue.size())
queue.offer((message, consumer)) match {
case QueueOfferResult.Enqueued =>
metricsRegistry.inputQueueAddEventCounter.increment()
case QueueOfferResult.Dropped =>
metricsRegistry.inputQueueDropEventCounter.increment()
consumer.nack()
logger.warn(s"Buffer is full, message nacked. Pubsub should retry don't panic. If this happens too often, we should also tweak the buffer size or the autoscaler.")
case QueueOfferResult.Failure(ex) =>
metricsRegistry.inputQueueDropEventCounter.increment()
consumer.nack()
logger.error(s"Failed to offer message with id=${message.getMessageId()}", ex)
case QueueOfferResult.QueueClosed =>
logger.error("Destination Queue closed. Something went terribly wrong. Shutting down the jvm.")
consumer.nack()
mat.shutdown()
sys.exit(1)
}
}
}
val subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId)
val subscriber = Subscriber.newBuilder(subscriptionName, receiver).setParallelPullCount(pullParallelism).build
subscriber.startAsync().awaitRunning()
source.withAttributes(attr)
}
}
}
【讨论】：