package TestKafka;
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.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class TestProducer {
public static void runProducer(int i) {
Properties props = new Properties();
props.put("bootstrap.servers", "10.190.35.131:9092");
props.put("acks", "all");
props.put("retries ", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i) + "!!"));
producer.close();
}
public static void runConsumer() {
KafkaConsumer<String, String> consumer;
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.190.35.131:9092"); //10.190.35.131:9092 10.86.45.40:9092
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); //"earliest" else "latest"
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer_group_test");
consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList("test")); // log.sweet.request.trace.production
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100); //Long.MAX_VALUE
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> record : partitionRecords) {
System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
System.out.println("");
}
}
consumer.commitSync();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
consumer.close();
}
}
public static void main(String[] args) throws Exception {
Thread thread = new Thread() {
public void run() {
int i = 0;
while (true) {
i++;
runProducer(i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
thread.start();
runConsumer();
// String s = String.format("this is number : %10.2f", 50.1);
// System.out.println(s);
}
}
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<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>com.geely.storm.test.producer</groupId>
<artifactId>TestProducer</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<kafka.client.version>0.10.0.0</kafka.client.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.client.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>${kafka.client.version}</version>
</dependency>
</dependencies>
</project>