java客户端连接kafka简单测试

java客户端连接kafka简单测试

本案例kafka版本是kafka_2.11-0.9.0.1，用java来实现kafka生产者、消费者的示例
在测试的过程中遇到的特别的问题以及解决办法，其他小问题就不一一列举了。

1 . 使用kafka-clients进行测试，maven依赖

<dependency>
         <groupId></groupId>
         <artifactId>kafka-clients</artifactId>
         <version>0.9.0.1</version>
    </dependency>

需要注意：kafka-clients的版本必须和kafka安装的版本一致
2 . Consumer测试类收不到消息的原因：

①. 先开启consumer测试类，等待poll数据。
②. 开启producer测试类，发送数据，并确保发送成功。
③. 正常情况下这样就可以收到数据了。

---

生产者示例

public class ProducerTest {
    private void execMsgSend() throws  Exception{
        Properties props = new Properties();
        ("", "阿里云外网Ip:9092");
        ("acks", "1");
        ("retries", 0);
        ("", 16384);
        ("", "");
        ("", "");
        Producer<String, String> procuder = new KafkaProducer<String, String>(props);

        String topic = "test";
        for (int i = 1; i <= 10; i++) {
            String value = " this is another message_" + i;
            ProducerRecord<String,String> record = new ProducerRecord<String, String>(topic,i+"",value);
            (record,new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println("message send to partition " + () + ", offset: " + ());
                }
            });
            System.out.println(i+" ----   success");
            (1000);
        }
        System.out.println("send message over.");
        ();
    }
    public static void main(String[] args) throws  Exception{
        ProducerTest test1 = new ProducerTest();
        ();
    }
}

---

消费者示例

public class Consumer {
    public static void main(String[] s){  
        Properties props = new Properties();
        ("", "阿里云外网Ip:9092");   
        ("", "1");
        ("", "true");
        ("", "1000");
        ("", "30000");
        ("", "");
        ("", "");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        (("test"));
        while (true) {
            System.out.println("poll start...");
            ConsumerRecords<String, String> records = (100);
            int count = ();
            System.out.println("the numbers of topic:" + count);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s", (), (), record.value());
        }
   }  
}