Kafka初识

bin/kafka-add-partitions.sh --topic test --partition 2  --zookeeper  192.168.197.170:2181,192.168.197.171:2181

public classClusterProducer extends Thread {
  private static final Log log =LogFactory.getLog(ClusterProducer.class);

  public void sendData() {
    Random rnd = new Random();
    Properties props =PropertiesParser.getProperties(PropertiesSettings.PRODUCER_FILE_NAME);
    if (props == null) {
      log.error("can't loadspecified file " + PropertiesSettings.PRODUCER_FILE_NAME);
       return;
    }
    //set the producer configurationproperties
    ProducerConfig config = newProducerConfig(props);
    Producer<String, String> producer= new Producer<String, String>(config);

    //Send the data
    KeyedMessage<String, String> data;
    int count = 0;
    String url = "jdbc:inceptor://127.0.0.1:1433;DatabaseName=tmp";  
    String username = "sa";  
    String password = "";  

    Connection conn = DriverManager.getConnection(url, username, password);  
    Statement stmt = conn.createStatement();  
    ResultSet rs = stmt.executeQuery("select hphm from tfc_pass");  
    while (rs.next()) {  
    data = new KeyedMessage<String,String>("tfc_pass", count+'', rs.getString("hphm") );
    producer.send(data);
    count++;
    } 

    stmt.close();  
    conn.close();  
    producer.close();
  }

  public void run() {
    sendData();
  }

  public static void main(String[] args) {
    new ClusterProducer().sendData();
  }
}

public class Consumerextends Thread {
  private static final Log log =LogFactory.getLog(Consumer.class);
  private final ConsumerConnector consumer;
  private final String topic;

  public Consumer(String topic) {
    consumer =kafka.consumer.Consumer.createJavaConsumerConnector(
        createConsumerConfig());
    this.topic = topic;
  }

  private static ConsumerConfigcreateConsumerConfig() {
    Properties props = new Properties();
     props.put("zookeeper.connect", KafkaProperties.zkConnect);
    props.put("group.id",KafkaProperties.groupId);
     props.put("zookeeper.session.timeout.ms", "400");
     props.put("zookeeper.sync.time.ms", "200");
     props.put("auto.commit.interval.ms", "1000");

    return new ConsumerConfig(props);

  }

  public void run() {
    String url = "jdbc:inceptor://127.0.0.1:1433;DatabaseName=tmp";  
    String username = "sa";  
    String password = ""; 
    Connection conn = DriverManager.getConnection(url, username, password);  
    Statement stmt = conn.createStatement();  

    Map<String, Integer>topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, newInteger(1));
    Map<String,List<KafkaStream<byte[], byte[]>>> consumerMap =consumer.createMessageStreams(topicCountMap);
    KafkaStream<byte[], byte[]>stream = consumerMap.get(topic).get(0);
    ConsumerIterator<byte[], byte[]>it = stream.iterator();
    while (it.hasNext()) {
      String val = new String(it.next().message());
      stmt.executeUpdate("insert into blacklist values('"+val+"')") ;
    }

    stmt.close();  
    conn.close();  
  }

  public static void main(String[] args) {
    Consumer client = new Consumer("cluster_statistics_topic");
    client.run();
  }
}