KafkaConsumer.java 文件源码

private void setUpConsumer( Map<String, Integer> topicMap, MessageHandler<?> handler, Properties consumerProps ) {
    _executors = new HashMap<String, ExecutorService>();
    _topicConsumers = new HashMap<String, ConsumerConnector>();

    for ( String topic : topicMap.keySet() ) {
        String normalizedTopic = topic.replace( ".", "_" );
        String normalizedConsumerGroupId = getGroupId( consumerProps.getProperty( "group.id" ), normalizedTopic );
        consumerProps.setProperty( "group.id", normalizedConsumerGroupId );
        LOG.warn( "Consuming topic '" + topic + "' with group.id '" + normalizedConsumerGroupId + "'" );
        LOG.warn( consumerProps.toString() );
        ConsumerConfig topicConfig = new ConsumerConfig( consumerProps );
        _topicConsumers.put( topic, kafka.consumer.Consumer.createJavaConsumerConnector( topicConfig ) );
    }
    _topicMap = topicMap;
    _handler = handler;
}