Kafka制作人(0.8.2)-经纪人倒下时如何追踪？

Question

感谢你在这个问题上的帮助。

我使用的是Kafka 0.8.2

这是我写的制作人代码。

public class cvsKafkaProducerAck {
public static void main(String args[]) {


    String topic = "test_topic";
    String msg = "test...22";

    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.RETRIES_CONFIG, "1");
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    //props.put(ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG , "1");
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1);
    props.put(ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG, true);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

    KafkaProducer<byte[], byte[]> m_kafkaProducer = new KafkaProducer<byte[], byte[]>(props);       
    ProducerRecord<byte[], byte[]> prMessage = new ProducerRecord<byte[],byte[]>(topic, msg.getBytes());

    try
    {
        RecordMetadata metadata = m_kafkaProducer.send(prMessage).get();
        System.out.println("The offset of the record we just sent is: " + metadata.offset());
    }
    catch(Exception e)
    {
        System.out.println(e.getMessage());
    }

    m_kafkaProducer.close();

}

Answer 1

我们有一个使用node-zookeeper-client库的Kafka代理监控系统。

代理在Zookeeper节点/Brokers/ids上注册。使用client.getChildren（）函数，我们不断地监视该路径，查找正在添加或删除的代理。client.getChildren()函数可以在Zookeeper节点上放置一个观察者，该观察者将在添加或删除子节点时激发一个事件。

我们的节点应用程序在启动时从/brokers/ids路径获取已知的代理，然后连续监视路径的更改，在添加或删除代理时发出警报。这是通过使用lodash库将当前代理的数组与上次检索到的代理列表的数组进行比较来完成的，例如。

//this code is fired when the /brokers/ids/ path changes (i.e. a child node is added or removed)

//currentBrokers = list of brokers retrieved when the app started
//newBrokerList = latest list of brokers retrieved from client.getChildren()
var brokersChanged = _.difference(currentBrokers, newBrokerList); 

  //if 'brokersChanged' contains any values alert that
  //these brokers have become unregistered with zookeeper
  if(brokersChanged.length > 0){
    _.each(brokersChanged, function(broker){
      var indexToRemove = _.findIndex(currentBrokers, function(existingBroker) {
        if(existingBroker == broker){
          return existingBroker;
        }
      });

      currentBrokers.splice(indexToRemove,1);
    });    
  }
  else {
    //if 'brokersChanged' is empty a new broker has been registered
    //with zookeeper
    var newBroker = _.difference(data, currentBrokers);
    _.each(newBroker, function(broker){
      currentBrokers.push(broker);
    });
  }

//currentBrokers will now contain the updated list of brokers