java操作kafka读写操作

1，前提，kafka 的 server.properties里面开通了

listeners=PLAINTEXT://192.168.137.141:9092
advertised.listeners=PLAINTEXT://192.168.137.141:9092

2，防火墙

systemctl status firewalld
systemctl stop firewalld

3，java代码

生产者

import java.text.SimpleDateFormat;
import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

public class Producer123 {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Producer123 aa1=new Producer123();
		for(int i=0;i<50;i++) {
			aa1.send1();
			System.out.println(i);
			try {
				Thread.sleep(300);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	public void send1() {
		String out = "";
		String topic, msg;
		Properties properties = new Properties();
		KafkaProducer<String, String> kafkaProducer;
		topic = "CHINA";
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.141:9092");
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		kafkaProducer = new KafkaProducer<String, String>(properties);
		Date dNow = new Date( );
		SimpleDateFormat ft = new SimpleDateFormat ("E yyyy.MM.dd 'at' hh:mm:ss a zzz");
		msg = "现在是" + ft.format(dNow);
		System.out.println(msg);
		ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, msg);
		kafkaProducer.send(producerRecord);
		kafkaProducer.close();

//	return out;
	}

}

4，消费者

import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class Main1 {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		System.out.println("hello");
		Main1 m1=new Main1();
		m1.consumer1();

	}
	public String consumer1() {
		System.out.println("starting......");
		String out = "";
		String topic;
		Properties properties = new Properties();
		KafkaConsumer<String, String> kafkaConsumer;
		topic = "CHINA";
		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.141:9092");
		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "lowLevel");
		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		kafkaConsumer = new KafkaConsumer<String, String>(properties);
		kafkaConsumer.subscribe(Collections.singletonList(topic));
		int i=0;
		while(i<10){
			i++;
			System.out.println("aaa:"+String.valueOf(i));
		    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(3000);//3000代表每3秒读一次
		    System.out.println(consumerRecords.count());
		    if(consumerRecords.count() > 0){
		        for (ConsumerRecord<String, String> consumerRecord : consumerRecords){
		            out = consumerRecord.value();
		            System.out.println("receive data: "+out);
		        }
//		        break;
		    }
		}
		return out;	
	}

}

5，依赖项，版本最好和kafka服务端保持一致

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>test-kafka</groupId>
  <artifactId>test-kafka</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>test-kafka</name>
  <description>test-kafka</description>
  <dependencies>
	  	<dependency>
		    <groupId>org.apache.kafka</groupId>
    		<artifactId>kafka-clients</artifactId>
    		<version>2.8.0</version>
		</dependency>
  </dependencies>
  <build>
       <plugins>
           <plugin>
               <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-compiler-plugin</artifactId>
               <configuration>
					       <source>1.8</source>
                           <target>1.8</target>
               </configuration>
           </plugin>
       </plugins>
   </build>
</project>