kafka之java编程模型

package com.ganglia.kafka;

import java.util.Date;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
   
public class ProducerTest2 {   
     
        public static void main(String[] args) {   
            Properties props = new Properties();   
            props.setProperty("metadata.broker.list","bfdbjc1:9092,test1:9092,test2:9092");   
            props.setProperty("serializer.class","kafka.serializer.StringEncoder");   
            ProducerConfig config = new ProducerConfig(props);   
            Producer<String, String> producer = new Producer<String, String>(config);   
            try {   
                int i =1; 
                while(true){ 
                    i++;
                    String text = new StringBuffer((i+"")).reverse()+":test-kafka_"+args[0]+"_"+i;
                    KeyedMessage<String, String> data = new KeyedMessage<String, String>("test",text);   
                    producer.send(data);   
                    Thread.sleep(100);
                    System.out.println(DateUtil.fmtDateToYMDHMS(new Date())+"\t"+text);
                } 
            } catch (Exception e) {   
                e.printStackTrace();   
            }   
            producer.close();   
        }   
}
package com.ganglia.kafka;

import java.util.Date