如何基于Actor可用性，通过Akka Kafka Connector控制Kafka消息拉取

一、先排查Kafka分区分配问题 #

当前Consumer2处于空闲状态，首先要确认目标topic的分区数量是否≥消费者实例数（2）。Kafka消费者组会将分区均匀分配给组内消费者，若分区数少于消费者数，部分消费者会无分区可消费，直接导致空闲。

解决方式：

• 调整topic的分区数至≥2，确保两个消费者都能分到对应分区，获得拉取消息的权限。

二、修复背压传递问题，绑定Actor池状态到Kafka拉取 #

你当前使用的Sink.actorRefWithBackpressure仅针对单个Actor生效，无法感知Actor池的整体空闲状态，导致背压无法传递到Kafka Source。需调整流的处理逻辑，将Actor池的并发处理能力与流的背压机制绑定：

1. 改用Actor池管理处理Actor #

创建大小为5的Actor池（以RoundRobinPool为例），替代单个Actor接收消息：

// 在ActorSystem中初始化Actor池
ActorRef processingPool = system.actorOf(
    new RoundRobinPool(5).props(Props.create(ProcessingActor.class)),
    "consumer-processing-pool"
);

2. 用mapAsyncUnordered替代Sink.actorRefWithBackpressure #

将流的并发处理数设置为Actor池的大小（5），通过ask模式等待Actor处理完成的确认，让流的背压直接关联Actor池的空闲状态：

public void createAndRegisterConsumer(String groupId, ActorRef actorPool, SourceType sourceType, String... topics) {
    RestartSource.onFailuresWithBackoff(Duration.ofSeconds(3), Duration.ofSeconds(20), 0.2,
            () -> createRawConsumer(groupId, sourceType, topics).mapMaterializedValue(c -> {
                topicMapping.put(topicKey(topics), c);
                return c;
            }))
            .map(ConsumerRecord::value)
            // 并发度与Actor池大小一致，最多同时处理5条消息
            .mapAsyncUnordered(5, message -> {
                // 向Actor池发送消息，等待处理完成的确认
                return AskPattern.ask(actorPool, message, Duration.ofSeconds(10))
                        .thenApply(ack -> null); // 仅触发背压继续，无需返回结果
            })
            .runWith(Sink.ignore(), materializer);
}

3. 调整ProcessingActor逻辑，处理完消息后回复确认 #

确保处理Actor在完成业务逻辑后，向发送方回复确认消息（比如Ack.INSTANCE），触发流继续拉取下一条消息：

public class ProcessingActor extends AbstractActor {
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(String.class, message -> {
                    // 执行消息处理逻辑
                    processMessage(message);
                    // 回复确认，告知流可以继续拉取
                    getSender().tell(Ack.INSTANCE, getSelf());
                })
                .build();
    }

    private void processMessage(String message) {
        // 你的业务处理代码
    }
}

4. 优化Kafka Consumer配置，增强背压效果 #

调整消费者配置，让Kafka消费者在没有下游需求时减少主动拉取，避免无意义的消息缓存：

private ConsumerSettings<byte[], String> getConsumerSettings(String groupId, KafkaConfig kafkaConfig, boolean isPlain) {
    ConsumerSettings<byte[], String> settings = ConsumerSettings.create(system, new ByteArrayDeserializer(), new StringDeserializer())
            .withBootstrapServers(kafkaConfig.getBootstrapServers())
            .withGroupId(groupId)
            // 配合背压，设置拉取的最小字节数和最大等待时间
            .withProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1")
            .withProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500")
            // 单次拉取的最大记录数匹配Actor池大小
            .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");

    // 原有其他配置...
    return settings;
}

三、原理说明 #

• mapAsyncUnordered(5, ...)会限制流中同时处理的消息数为5，与Actor池的大小完全匹配。当所有5个Actor都处于忙碌状态时，流会暂停从Kafka Source拉取新消息，直到有Actor完成处理并回复确认。
• Kafka的plainSource和atMostOnceSource本身支持背压，当下游流暂停需求时，消费者会停止向Kafka集群请求新消息，彻底避免无意义的消息拉取和缓存。

内容的提问来源于stack exchange，提问作者user3871913