logstash kafka多台机器取数据,只有一个实例消费【logstash2.3.4】

2016-10-31 19:04:10来源:CSDN作者:fsz9065人点击

第七城市

描述:首先kafka的partition数量为6,多台logstash服务机器的topic相同并且groupId也相同。当分别启动机器时,只有一台机器在消费kafka数据,并且总是最后启动的那台在消费。修改了zk时间等各种kafka配置均没有效果,最后不得不看ruby源码了。感觉kakfa的代码写的还是有点问题的。

首先介绍logstash的input-kafka模块的代码结构

kafka.rb -- logstash-2.3.4/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-2.0.8/lib/logstash/inputs/group.rb -- logstash-2.3.4/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka

主要的代码逻辑集中在这两个ruby文件上,kafka.rb主要接收配置参数,并调用group.rb对kafka数据进行消费,这个过程是调用的kafka java api,大体流程于java代码类似。
仔细读下去会发现group.rb的代码有点问题:

  def run(a_num_threads, a_queue)    begin      if @reset_beginning == 'from-beginning'        Java::kafka::utils::ZkUtils.maybeDeletePath(@zk_connect, "/consumers/#{@group_id}")      end      @consumer = Java::kafka::consumer::Consumer.createJavaConsumerConnector(create_consumer_config)    rescue ZkException => e      raise KafkaError.new(e), "Got ZkException: #{e}"    end    thread_value = a_num_threads.to_java Java::int    streams = get_streams(thread_value)    @executor = Executors.newFixedThreadPool(a_num_threads)    @executor_submit = @executor.java_method(:submit, [Java::JavaLang::Runnable.java_class])    thread_number = 0    streams.each do |stream|      @executor_submit.call(Kafka::Consumer.new(stream, thread_number, a_queue, @consumer_restart_on_error, @consumer_restart_sleep_ms))      thread_number += 1    end    @running = true  end

在定义的run方法中,

if @reset_beginning == 'from-beginning'        Java::kafka::utils::ZkUtils.maybeDeletePath(@zk_connect, "/consumers/#{@group_id}")      end

主要含义是删除zk目录下,该groupId的信息。就因为有这句话,原来logstash实例的消费信息就会被移除了,所以会出现上述现象(具体这样做的含义,因为时间原因没有过多的调研)

————–这里是解决方法———————
logstash-2.3.4/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka 目录下,打开group.rb文件,找到代码片,用#注释掉即可

#if @reset_beginning == 'from-beginning'      #Java::kafka::utils::ZkUtils.maybeDeletePath(@zk_connect, "/consumers/#{@group_id}") #end
第七城市

最新文章

123

最新摄影

微信扫一扫

第七城市微信公众平台