参考: https://www.orchome.com/451Kafka集群的安装见上文本文介绍使用Java API通过kafka发送和接收消息。1. kafka客户端依赖dependency groupIdorg.apache.kafka/groupId artifactIdkafka_2.11/artifactId version1.0.1/version /dependency dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version1.0.1/version /dependency2 Kafka消息生产者APIpackage kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; public class ProducerDemo { private static final String MY_TOPIC my-topic; public static void main(String[] args) { Properties properties new Properties(); // Kafka 服务器地址 properties.put(bootstrap.servers, 127.0.0.1:9092,127.0.0.1:9093); // 消息应答机制 properties.put(acks, all); // 如果请求失败生产者会自动重试我们指定是0次如果启用重试则会有重复消息的可能性 properties.put(retries, 0); properties.put(batch.size, 16384); // 默认缓冲可立即发送即便缓冲空间还没有满但是如果你想减少请求的数量可以设置linger.ms大于0 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); // 创建并发送消息 try (ProducerString, String producer new KafkaProducer(properties)) { for (int i 0; i 100; i) { String msg Message-index- i; producer.send(new ProducerRecord(MY_TOPIC, msg)); System.out.println(Sent: msg); } } } }消息发送结果:3. Kafka消息消费者APIpackage kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Collections; import java.util.Properties; public class ConsumerDemo { private static final String MY_TOPIC my-topic; public static void main(String[] args) { Properties properties new Properties(); // kafka 服务器地址 properties.put(bootstrap.servers, 127.0.0.1:9092); // 当前消费者所在的consumer group 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); // 停止心跳的时间超过session.timeout.ms,那么就会认为是故障的它的分区将被分配到别的进程 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); // 订阅 my-topic 主题的消息 KafkaConsumerString, String consumer new KafkaConsumer(properties); consumer.subscribe(Collections.singletonList(MY_TOPIC)); // 不停的获取消息并消费 while (true) { ConsumerRecordsString, String records consumer.poll(1000); System.out.println(records count: records.count()); for (ConsumerRecordString, String record : records) { System.out.printf(offset %d, key %s, value %s, record.offset(), record.key(), record.value()); System.out.println(); } } } }消息消费的结果如下
参考: https://www.orchome.com/451Kafka集群的安装见上文本文介绍使用Java API通过kafka发送和接收消息。1. kafka客户端依赖dependency groupIdorg.apache.kafka/groupId artifactIdkafka_2.11/artifactId version1.0.1/version /dependency dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version1.0.1/version /dependency2 Kafka消息生产者APIpackage kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; public class ProducerDemo { private static final String MY_TOPIC my-topic; public static void main(String[] args) { Properties properties new Properties(); // Kafka 服务器地址 properties.put(bootstrap.servers, 127.0.0.1:9092,127.0.0.1:9093); // 消息应答机制 properties.put(acks, all); // 如果请求失败生产者会自动重试我们指定是0次如果启用重试则会有重复消息的可能性 properties.put(retries, 0); properties.put(batch.size, 16384); // 默认缓冲可立即发送即便缓冲空间还没有满但是如果你想减少请求的数量可以设置linger.ms大于0 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); // 创建并发送消息 try (ProducerString, String producer new KafkaProducer(properties)) { for (int i 0; i 100; i) { String msg Message-index- i; producer.send(new ProducerRecord(MY_TOPIC, msg)); System.out.println(Sent: msg); } } } }消息发送结果:3. Kafka消息消费者APIpackage kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Collections; import java.util.Properties; public class ConsumerDemo { private static final String MY_TOPIC my-topic; public static void main(String[] args) { Properties properties new Properties(); // kafka 服务器地址 properties.put(bootstrap.servers, 127.0.0.1:9092); // 当前消费者所在的consumer group 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); // 停止心跳的时间超过session.timeout.ms,那么就会认为是故障的它的分区将被分配到别的进程 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); // 订阅 my-topic 主题的消息 KafkaConsumerString, String consumer new KafkaConsumer(properties); consumer.subscribe(Collections.singletonList(MY_TOPIC)); // 不停的获取消息并消费 while (true) { ConsumerRecordsString, String records consumer.poll(1000); System.out.println(records count: records.count()); for (ConsumerRecordString, String record : records) { System.out.printf(offset %d, key %s, value %s, record.offset(), record.key(), record.value()); System.out.println(); } } } }消息消费的结果如下
