kafka的helloworld程序

原创
2017/06/20 11:02
阅读数 334

POM 

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.0</version>
        </dependency>

 

生产者:

public class KafkaProducerTest {
    private static final Logger LOG = LoggerFactory.getLogger(Mqtt2Kafka.class);

    public static void main(String[] avgs){
        LOG.debug("开始");

        Producer<String, String> producer = new KafkaProducer<>(KafkaParams.getSer9ParamsSender());
        int i=1;
        while (i<=10_0000){
            producer.send(new ProducerRecord<>("test2", 0, null, Integer.toString(i)));
            if (i%1000==0)
                producer.flush();
            i++;
        }
        producer.flush();
        producer.close();
        LOG.debug("结束");
    }
}

消费者:

public class KafkaConsumerTest {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerTest.class);

    public static void main(String[] avgs) {
        LOG.debug("开始");
        Consumer<String, String> consumer = new KafkaConsumer<>(KafkaParams.getSer9ParamsSender(), new StringDeserializer(), new StringDeserializer());

        consumer.subscribe(Arrays.asList("test2"));

        while (true){

            ConsumerRecords<String, String> poll = consumer.poll(20);
            Iterator<ConsumerRecord<String, String>> iterator = poll.iterator();

            while (iterator.hasNext()){
                ConsumerRecord<String, String> next = iterator.next();
                System.out.println("key:"+ next.key() + "---value:"+ next.value());
            }
        }

    }

}

 

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部