技术:java8 + kafka client
运行环境:IDEA 15.2 + jdk8 + windows 10
demo功能:提供一个java 操作kafka最基本demo
org.apache.kafka
kafka-clients
0.11.0.0
public class KafkaProducerSender {
public static Properties properties = new Properties();
static {
properties.put("bootstrap.servers", "10.32.8.5:9092,10.32.0.116:9092,10.32.8.10:9092");
properties.put("acks", "all");
properties.put("retries", 0);
properties.put("batch.size", 16384);
properties.put("linger.ms", 1);
properties.put("buffer.memory", 33554432);
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
}
public static String topic = "HelloWorld";
public static void sendDemo() {
System.out.println("start produce...");
Producer<String, String> producer = null;
try {
producer = new KafkaProducer<>(properties);
for (int i = 0; i < 100; i++) {
String msg = "Message " + i;
producer.send(new ProducerRecord<>(topic, msg));
System.out.println("Sent:" + msg);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.close();
}
}
}
public class KafkaConsumerReciever {
public static void recieve() {
Properties properties = new Properties();
properties.put("bootstrap.servers", KafkaProducerSender.properties.get("bootstrap.servers"));
properties.put("group.id", "group-1");
properties.put("enable.auto.commit", "true");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "30000");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
kafkaConsumer.subscribe(Arrays.asList(KafkaProducerSender.topic));
while (true) {
System.out.println("i'am alive...");
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
if (records.isEmpty()) {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, value = %s", record.offset(), record.value());
System.out.println();
}
}
}
}
public class Application {
public static void main(String[] args) {
KafkaProducerSender.sendDemo();
KafkaConsumerReciever.recieve();
}
}
欢迎添加微信,互相学习↑↑↑ -_-
白老虎
programming is not only to solve problems, ways to think
grafana 级连 菜单 templating (variables) 配置
rocketmq 集群搭建 (2master + 2slave + 2namesrv)
AI 机器人 抓取 微信 聊天中的 百度网盘 分享地址和密码