Kafka Streaming 不适用于多个实例答案

【问题标题】：Kafka Streaming not working with multiple instancesKafka Streaming 不适用于多个实例
【发布时间】：2017-09-12 10:18:44
【问题描述】：

当我运行 Kafka Streams 应用程序的多个实例时，只有第一个实例正确接收消息。但是，如果我启动新实例，它们不会收到任何消息。

有解决此问题的建议吗？

这是我的 Kafka 流媒体应用程序

package test.kafkastream;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;

public class Main {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount-processor");
        //props.put(ConsumerConfig.GROUP_ID_CONFIG, "streams-wordcount-processor");

        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.38:45983,192.168.2.112:45635,192.168.2.116:39571");
        //props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);


        // setting offset reset to earliest so that we can re-run the demo code
        // with the same pre-loaded data
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        TopologyBuilder builder = new TopologyBuilder();

        builder.addSource("Source", "topic6");

        builder.addProcessor("Process", new ProcessMessage(), "Source");

        KafkaStreams streams = new KafkaStreams(builder, props);
        streams.start();
    }

}

这是我的制片人

package test.kafkamesos;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;

public class Producer {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Map<String, Object> producerConfig = new HashMap<String, Object>();
        producerConfig.put("bootstrap.servers", "192.168.2.38:45983,192.168.2.112:45635,192.168.2.116:39571");
        //producerConfig.put("bootstrap.servers", "localhost:9092");

        // optional:
        producerConfig.put("metadata.fetch.timeout.ms", "3000");
        producerConfig.put("request.timeout.ms", "3000");
        // ... other options:
        // http://kafka.apache.org/documentation.html#producerconfigs
        ByteArraySerializer serializer = new ByteArraySerializer();
        KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[], byte[]>(producerConfig, serializer,
                serializer);

        int i = 0;
        while (true) {
            String message = "{data:success,g:" + i + "}";
            ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>("topic6", message.getBytes());
            kafkaProducer.send(record).get();
            System.out.println("sending " + message);
            Thread.sleep(1000);
            i++;
        }
    }
}

还有我的 Dockerfile

FROM openjdk:8-jre
COPY ./target/*-with-dependencies.jar /jars/service-jar.jar
CMD java -cp /jars/service-jar.jar test.kafkastream.Main

【问题讨论】：

我只能假设这是因为您添加了指向代码的链接，而不是代码本身。不过，如果是这个原因，还是应该说明的。
@AleksandarStojadinovic 谢谢。我现在将添加代码....

标签： java apache-kafka kafka-consumer-api apache-kafka-streams

【解决方案1】：

我相信您遇到了这个问题，因为 Kafka 代理只为您正在使用的主题配置了一个分区 (topic6)。来自 Confluent 博客：

例如，如果您的应用程序从一个包含 10 分区，那么您最多可以运行 10 个应用程序实例（请注意，您可以运行更多实例，但这些实例将处于空闲状态）。在总之，主题分区的数量是您的 Streams API 应用程序的并行性，因此对于运行应用程序的实例。

来源：https://www.confluent.io/blog/elastic-scaling-in-kafka-streams/

【讨论】：