当发送给分片演员时，对ReactiveKafka的背压

我已经编写了一个Akka应用程序，该应用程序从Kafka获取输入，然后使用分片演员处理数据并输出到Kafka。当发送给分片演员时，对ReactiveKafka的背压

但在某些场合分片区域不能处理负载，我也得到：

你或许应该实行流量控制，以避免水浸远程连接。

如何在此链/流中实施背压？

卡夫卡消费 - >共享演员 - >卡夫卡生产者

从代码一些片断：

ReactiveKafka kafka = new ReactiveKafka(); 

Subscriber subscriber = kafka.publish(pp, system); 

ActorRef kafkaWriterActor = (ActorRef) Source.actorRef(10000, OverflowStrategy.dropHead()) 
       .map(ix -> KeyValueProducerMessage.apply(Integer.toString(ix.hashCode()), ix)) 
       .to(Sink.fromSubscriber(subscriber)) 
       .run(materializer); 

ConsumerProperties cp = new PropertiesBuilder.Consumer(brokerList, intopic, consumergroup, new ByteArrayDeserializer(), new NgMsgDecoder()) 
         .build().consumerTimeoutMs(5000).commitInterval(Duration.create(60, TimeUnit.SECONDS)).readFromEndOfStream(); 

Publisher<ConsumerRecord<byte[], StreamEvent>> publisher = kafka.consume(cp,system); 

ActorRef streamActor = ClusterSharding.get(system).start("StreamActor", 
       Props.create(StreamActor.class, synctime), ClusterShardingSettings.create(system), messageExtractor); 

shardRegionTypenames.add("StreamActor"); 


Source.fromPublisher(publisher)     
       .runWith(Sink.foreach(msg -> {      
        streamActor.tell(msg.value(),ActorRef.noSender()); 
       }), materializer);

来源

2017-04-10 Hubert

我是Akka框架的新手。但是您可以尝试使用Akka Streams来照顾背压技术。 – Advika

也许你可以考虑并行的主题为分区（如果适用），并与per-消费者打造通过调整ConsumerWithPerPartitionBackpressure中的this example与您的演员使用mapAsync and ask进行整合来分区背压。

来源

2017-04-11 21:58:09

当发送给分片演员时，对ReactiveKafka的背压

回答

相关问题