gistfile1.java

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.javaapi.producer.ProducerData;
import kafka.producer.ProducerConfig;

public class DCSProducer {

	public static void main(String args[]) {
		Properties props = new Properties();
		props.put("zk.connect", "use-rabbitmq01.ec2.demdex.com:2181");
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("producer.type", "async");
		// props.put("queue.enqueueTimeout.ms", "2");
		ProducerConfig config = new ProducerConfig(props);
		Producer<String, String> producer = new Producer<String, String>(config);
		try {
			for (int i = 0; i < 200000; i++) {
				ProducerData<String, String> data = new ProducerData<String, String>("aam.segments", "test-message sdflkjdj sdfljslkdjf sdflkjsdlkfjs sdfjsdlkfj sdfljsdlkfj sdfjsdlkfj sfjsdlkfjsldjf sdfjsdlfkjsdlfj sdfjsdlfkjlksdjf sdfljsdlfkjsdflj sdfjsdlfkjsdfw gedfgdfg dfdfgdfg dsdfsdf sdfsdfsdf sdfsdf sdfsdf sdfsdfsf sdfsdfdsf abcde abcde sfkldsf sdfsdf sdfsdf sdfsdf adsd asdasd asd asdasd asdasd asdasd asdasd asdasd asdasd asdasd sdfsdf sdfsdfs sdfsdf sdfsdf sdfsdf sdfsdf sfsdf sdfsdf sdf sdf sdf weerewr werwerwer werwerwer werwerwerw sf sdfsdfsdf dsf sdf sdf sdf sdf sdf s df sdfs df sdf" + i);
				producer.send(data);
				// System.out.println("sent " + i);
			}
		} finally {
			System.out.println("close");
			producer.close();
		}
	}
}