java - 生产者不能将消息分发给多个消费者？

问题描述

我正在尝试使用带有--scale标志的 docker-compose 自动扩展 java spring 框架 kafka 消费者，并在 docker-compose.yaml 中转发端口，例如“8070-8072:8070”。所以当我触发发布消息的端点时，它做得很好。但在消费者方面，只有 1 个消费者消费所有消息。我有 3 个具有相同组 ID 和不同客户端 ID 的消费者。我想要的是分布式消息传递。我阅读了一些关于分区的论文，并查看了我的日志。似乎只有1个分区。是这个原因吗？我怎样才能做到这一点？我将添加日志、消费者配置、发布者配置和 docker-compose 文件。首先是日志。似乎只有 1 个中有 3 个有分区。

命令：

docker-compose up --build --scale nonwebapp=3 --scale webapp=3

docker-compose.yaml

kafka:
        image: confluentinc/cp-kafka:latest
        depends_on:
            - zookeeper
        ports:
            - 9092:9092
        environment:
            KAFKA_LOG_DIRS: "/kafka/kafka-logs"
            KAFKA_BROKER_ID: 1
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
            KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
            KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
        volumes:
            - /var/run/docker.sock:/var/run/docker.sock
        restart: always
            
    webapp:
        build: benchmark.web
        command: bash -c "while !</dev/tcp/kafka/29092; do >&2 echo 'Waiting for kafka to up...' sleep 1; done;"
        ports: 
            - "8070-8072:8070"
        volumes:
            - ./logs:/logs
            - ${GCP_KEY_PATH}:/root/keys/keyfile.json:ro
        depends_on:
          - kafka
        environment:
          SPRING_KAFKA_BOOTSTRAPSERVERS: kafka:29092
          TZ: "Asia/Istanbul"
          GOOGLE_APPLICATION_PROJECT_ID: apt-cubist-282712
          GOOGLE_APPLICATION_CREDENTIALS: /root/keys/keyfile.json
        restart: always
            
            
    nonwebapp:
        build: benchmark.nonweb
        command: bash -c "while !</dev/tcp/kafka/29092; do >&2 echo 'Waiting for kafka to up...' sleep 1; done;"
        depends_on:
          - kafka
        volumes:
          - ${GCP_KEY_PATH}:/root/keys/keyfile.json:ro
        ports:
          - "8060-8062:8060"
        environment:
          SPRING_KAFKA_BOOTSTRAPSERVERS: kafka:29092
          GOOGLE_APPLICATION_PROJECT_ID: apt-cubist-282712
          GOOGLE_APPLICATION_CREDENTIALS: /root/keys/keyfile.json
          TZ: "Asia/Istanbul"
        restart: always

生产者配置

@Bean
    ProducerFactory<String, byte[]> producerFactory(){
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaInContainerAdress);

    /*
        configProps.put(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaInLocal);
*/
        configProps.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class
        );
        configProps.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                ByteArraySerializer.class
        );
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String,byte[]> kafkaTemplate(){

        return new KafkaTemplate<>(producerFactory());
    }

消费者配置

 @Bean
    public ConsumerFactory<String, byte[]> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
       props.put(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaInContainerAdress);
       /*
        props.put(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaInLocal);
*/
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
                "org.apache.kafka.clients.consumer.RoundRobinAssignor");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, r.nextInt());
        props.put(
                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        props.put(
                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                ByteArrayDeserializer.class);
        props.put("group.id","topic_trial_consumers");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, byte[]>
    kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

标签： javaspring-bootdockerapache-kafkamessaging

似乎只有1个分区。是这个原因吗？

是的，如果只有一个分区，那么只有一个消费者（来自消费者组）可以消费它，其他消费者（来自同一组）即使启动也会处于空闲状态。

似乎只有 1 个中有 3 个有分区

从您的图像中，我可以看到topic_trial-0. 所以它是的topic_trial第一个分区。

增加编号。分区，例如 3 个，并以相同group.id的方式启动三个消费者，负载应该得到分布（每个分区一个）。

java - 生产者不能将消息分发给多个消费者？

问题描述

解决方案

推荐阅读