利用Java客户端简单操作kafka（读写操作）

曾经终败给现在 2023-06-30 09:14 11阅读 0赞

这里先简单地利用Java客户端向Kafka发送简单的消息，String类型的字符串。

首先，需要导入相关的包。我们选择Maven工程。相关的依赖如下：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>0.9.0.0</version>
    </dependency>

修改Kafka的配置文件：在kafka/config/server.properties中，修改hostname为ip地址（原来为localhost：）

（查看IP地址：ifconfig -a）

![20200116171431349.png][]

**（如果不修改，就会造成：在安装有Kafka的机器上可以正常读写消息，但是，在其他的机器上就nos，也不会报错误！）**

接下来：

**生产者：**

public class KafkaProducerTest {
        public static void main(String[] args) {
            Properties kafkaProps=new Properties();
            kafkaProps.put("bootstrap.servers","192.168.123.66:9092");
            kafkaProps.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
            kafkaProps.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    
            KafkaProducer producer=new KafkaProducer<String,String>(kafkaProps);
    
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "TestMessage 99");
    
            try {
                producer.send(record).get();
                System.out.println("成功发送消息到Kafka！");
            }catch (Exception e)
            {
                e.printStackTrace();
            }
    
        }
    }

**消费者：**

public class KafkaConsumerTest {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.123.66:9092");
            props.put("group.id", "CountryCounter");
            props.put("key.deserializer",  "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer",  "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
    
            //订阅主题
            //consumer.subscribe(Collections.singletonList("test"));//从分区的最新的偏移量开始
            //从分区的开头开始
             TopicPartition tp = new TopicPartition("test", 0);
            List<TopicPartition> list = new ArrayList<TopicPartition>();
             list.add(tp);
             consumer.assign(list);//需要指派主题分区列表
            consumer.seekToBeginning(tp);//从指定主题的特定分区开始
    
            //轮询
            try{
                while (true)
                {
                    ConsumerRecords<String, String> records = consumer.poll(100);
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.println("topic:"+record.topic()+",partition:"+record.partition()+",offset:"+record.offset()+
                                ",key:"+record.key()+",value:"+record.value());
                    }
                }
            }catch (Exception e)
            {
                e.printStackTrace();
            }finally {
                consumer.close();
                System.out.println("测试消费者！");
            }
        }
    }

提供有从分区的最新offset开始：就直接订阅就可以

consumer.subscribe(Collections.singletonList("test"));//从分区的最新的偏移量开始

从头开始读：

//从分区的开头开始
             TopicPartition tp = new TopicPartition("test", 0);
            List<TopicPartition> list = new ArrayList<TopicPartition>();
             list.add(tp);
             consumer.assign(list);//需要指派主题分区列表
            consumer.seekToBeginning(tp);//从指定主题的特定分区开始

这样简单的读写String类型的数据就ok了。

[20200116171431349.png]: https://img-blog.csdnimg.cn/20200116171431349.png