Kafka API8 producer

import java.util.Properties; import org.apache.log4j.Logger; import com.jd.Seckill.UdpRecv.Utils.Config; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig;   public class KafkaProduce_API8_1 extends Thread { Logger log = Logger.getLogger(KafkaProduce_API8_1.class); @Override public void run() { Handle(); } public void Handle() { try { int events = 100; Properties props = new Properties(); props.put("metadata.broker.list", Config.getM_strBrokerServers()); props.put("serializer.class", "kafka.serializer.StringEncoder"); props.put("request.required.acks", "1");   ProducerConfig config = new ProducerConfig(props);   Producer<String, String> producer = new Producer<String, String>(config);   for (int nEvents = 0; nEvents < events; nEvents++) { KeyedMessage<String, String> data = new KeyedMessage<>(Config.getM_strConsume8Topic(), Integer.toString(nEvents), Integer.toString(nEvents)); log.info("contents:" + nEvents); producer.send(data); } producer.close(); } catch (Exception e) { // TODO: handle exception log.error(e.toString()); } } }