package cu.kafka; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class ProducerDemo { public static void main(String[] args) throws Exception{ Properties props = new Properties(); props.put("zk.connect", "localhost:2181"); props.put("metadata.broker.list", "localhost:9092"); props.put("serializer.class", "kafka.serializer.StringEncoder"); ProducerConfig config = new ProducerConfig(props); Producer producer = new Producer(config); for(int i=1;i<=100000;i++){ producer.send(new KeyedMessage("test", "message: "+i)); } } } ![实验楼](https://dn-simplecloud.shiyanlou.com/1714651501903031420-wm) package cu.kafka; import java.util.*; import kafka.consumer.*; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata; public class ConsumerDemo { private static final String topic = "test"; private static final Integer threads = 1; public static void main(String[] args){ Properties props = new Properties(); props.put("zookeeper.connect", "localhost:2181"); props.put("group.id", "group1"); props.put("auto.offset.reset", "smallest"); ConsumerConfig config = new ConsumerConfig(props); ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config); Map topicCountMap = new HashMap(); topicCountMap.put(topic, 1); Map>> consumerMap = consumer.createMessageStreams(topicCountMap); List> streams = consumerMap.get(topic); for(final KafkaStream kafkaStream : streams){ new Thread(new Runnable() { @Override public void run() { for(MessageAndMetadata mm : kafkaStream){ String msg = new String(mm.message()); System.out.println(msg); } } }).start(); } } }

“Kafka快速上手教程”实验报告