这是正确的 Kafka Consumer Config - 在这个 Kafka 设置下吗？

Question

我的Kafka Producer 正在以大约 .. 350 mb 每 30 seconds.. 的速度生成消息。

Kafka Setup:

--> 1 Zookeeper instance

--> 3 Kafka Brokers

--> 1 Java Producer

--> 1 Java Consumer

这就是我创建Topic and broker partitions的方式：

bin/kafka-topics.sh --create --zookeeper 10.10.1.5:2181 --replication-factor 1 --partitions 8 --topic test

其余配置如下..

Producer Code:

KeyedMessage<String, byte[]> publishData = new KeyedMessage<String, byte[]>(this.topic, data);
producer.send(publishData);

这里，data 是 5000 长度 byte[]。

Producer Config:

batch.size = 200
producer.type = async
sflow-topic = test
connect.timeout.ms = 10000
request.required.acks = 0
zk.connect = 10.10.1.5:2181
serializer.class = kafka.serializer.DefaultEncoder
partitioner.class = kafka.producer.DefaultPartitioner
metadata.broker.list = 10.10.1.5:9092,10.10.1.6:9092,10.10.1.7:9092

我可以看到我的生产者工作得很好。问题在于消费者消费消息。即使消费者落后，我也看不到我的消息被消费（并最终处理并插入数据库）相同/相同的速度..另外，我对消费者进行了一些测试，发现并非所有消息都被我的消费者消费..不知道为什么:(

Consumer Code:

 public class FlowConsumer {
    private final String topic;
    private final ExecutorService threadPool;
    private final ConsumerConnector consumer;
    private static AppProperties appProperties;
    private final ExecutorService processDataThreadPool;

    public FlowConsumer() throws Exception {
        /**
         * Load properties configuration for flowLog4j.properties.
         */
        appProperties = AppProperties.loadConfiguration();

        /** Assign the flow-topic.. */
        this.topic = appProperties.getString(AppConstants.FLOW_TOPIC);
        logger.fatal("Topic : "+topic);

        /** Initialize the thread pool to consume kafka byte[] streams.. */
        this.threadPool = Executors.newFixedThreadPool(20);

        /** Initialize the thread pool for processing kafka byte[] messages.. */
        this.processDataThreadPool = Executors.newFixedThreadPool(100);

        /** Fetch the Consumer Config, by reading the Flow.properties file.. */
        this.consumer = kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerPropertyConfig.getConsumerConfig(appProperties));

        logger.fatal("Consumer : "+consumer);

        //new Thread(new Consumer()).start();
        threadPool.submit(new Consumer());
    }

    public void shutdown() {
        if (consumer != null) consumer.shutdown();
        if (threadPool != null) threadPool.shutdown();
        if (processDataThreadPool!= null) processDataThreadPool.shutdown();
    }

    private class Consumer implements Runnable {

        public Consumer() {
            logger.fatal("Started Consumer Thread!");
        }

        @Override
        public void run() {
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
            Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
            List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
            for (final KafkaStream<byte[], byte[]> kafkaStream : streams) {
                for (MessageAndMetadata<byte[], byte[]> messageAndMetadata : kafkaStream) {
                    processDataThreadPool.submit(new FlowServiceImpl(messageAndMetadata.message()));
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        FlowConsumer consumer = new FlowConsumer();

        /*try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {

        }
        consumer.shutdown();*/
    }
}

Consumer Config:

group.id = group1
flow-topic = test
auto.offset.reset = smallest
auto.commit.interval.ms = 2000
zookeeper.connect = 10.10.1.5:2181
zookeeper.sync.time.ms = 2000
zookeeper.session.timeout.ms = 2000
zookeeper.connection.timeout.ms = 6000

Question 1：

对于 3 个代理，我可以/应该创建多个 3 分区吗？我读到更多的分区意味着我可以为我的消费者添加更多的并行性？但是，如何通过在单个消费者上使用更多消费者线程？或者通过拥有 3 个消费者实例，每个拥有 1 个线程？

Question 2:

我的 Java 消费者配置代码是 correct/wrong 吗？

谁能告诉我我在这里做错了什么？

Answer 1

1. 您可以创建 3 个以上的分区，但不清楚您将获得什么加速；取决于您的网络、磁盘等。对于 N 个分区，您需要 N 个使用者线程在一台或多台机器上分别读取一个分区。

2. 您确定所有消息一开始都会进入 Kafka 吗？对于异步生产者，您需要 request.required.acks 为 -1 以获得最佳保证。有一个 kafka 日志转储工具，可用于检查代理上的数据。在消费者端，记录您阅读的消息的 Kafka 偏移量，以验证是否有跳过的消息。

Answer 2

首先，你可以在broker查看server.properties中名为log.dir的日志文件夹，你可以运行“tail -f”查看消息是否发送成功。

Answer 3

答案#1：是的，您可以为 3 个代理创建 8 个分区。不。 of brokers 对 no 没有任何限制。的分区。 例如，您有 8 个表正在获取数据并提交给主题。如果您的生产者以表名作为键向主题提交消息，并且您有 8 个分区，则很可能每个表消息都会发送到自己的分区。

通过匹配您的消费者数量等于分区数量意味着所有消费者并行读取到一个分区并消费消息。

按照设计，我会选择 8 个消费者实例，其中 1 个线程作为同一个消费者组运行。这样一个消费者实例将被分配给一个分区。如果您的消费者实例死了，那么您将只有一个分区数据落后于并非全部。

答案#2：在您的使用者代码中，您有大小为 20 的线程池读取 8 个分区，这意味着 12 个线程将保持空闲。 尝试使用相同的 group.id 并行运行 8 个实例并分享您的结果。

编辑：这对阅读也很有帮助

http://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster/