kafka-clients

添加依赖

org.apache.kafka

kafka-clients

2.5.0

消费者 Consumer

代码上总体可以分为三部分：消费者的配置消费者的配置在 org.apache.kafka.clients.consumer.ConsumerConfig 类中都有列举包括每个配置项的文档说明

创建消费者实例并订阅topic

消费消息

代码如下：// 1. 配置

Properties properties = new Properties();

//bootstrap.servers kafka集群地址 host1:port1,host2:port2 ....

properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

// key.deserializer 消息key序列化方式

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// value.deserializer 消息体序列化方式

properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// group.id 消费组id

properties.put(ConsumerConfig.GROUP_ID_CONFIG, "demo-group");

// enable.auto.commit 设置自动提交offset

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);

// auto.offset.reset

properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

// 2. 创建消费者实例并订阅topic

KafkaConsumer consumer = new KafkaConsumer<>(properties);

String[] topics = new String[]{"demo-topic"};

consumer.subscribe(Arrays.asList(topics));

// 3. 消费消息

while (true) {

ConsumerRecords records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord record : records) {

System.out.println(record);

}

}

生产者 Producer

生产者这块的代码基本上和消费者的结构一样，不同的是，producer 的发消息的部分：生产者的配置在 org.apache.kafka.clients.producer.ProducerConfig 类中也都有列举

创建生产者实例

发送消息到 topic异步发送消息 producer.send(new ProducerRecord<>("demo-topic", data))

同步发送消息 ，使用 Future.get() 阻塞接收

异步发送消息，回调的方式

整体代码如下// 1. 配置

Properties properties = new Properties();

// bootstrap.servers kafka集群地址 host1:port1,host2:port2 ....

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

// key.deserializer 消息key序列化方式

properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// value.deserializer 消息体序列化方式

properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 2. 创建生产者实例

KafkaProducer producer = new KafkaProducer<>(properties);

// 3. 发送消息

// 0 异步发送消息

for (int i = 0; i < 10; i++) {

String data = "async :" + i;

// 发送消息

producer.send(new ProducerRecord<>("demo-topic", data));

}

// 1 同步发送消息 调用get()阻塞返回结果

for (int i = 0; i < 10; i++) {

String data = "sync : " + i;

try {

// 发送消息

Future send = producer.send(new ProducerRecord<>("demo-topic", data));

RecordMetadata recordMetadata = send.get();

System.out.println(recordMetadata);

} catch (Exception e) {

e.printStackTrace();

}

}

// 2 异步发送消息 回调callback()

for (int i = 0; i < 10; i++) {

String data = "callback : " + i;

// 发送消息

producer.send(new ProducerRecord<>("demo-topic", data), new Callback() {

@Override

public void onCompletion(RecordMetadata metadata, Exception exception) {

// 发送消息的回调

if (exception != null) {

exception.printStackTrace();

} else {

System.out.println(metadata);

}

}

});

}

producer.close();

整合SpringBoot

添加依赖

org.springframework.boot

spring-boot-starter-parent

2.3.2.RELEASE

....

....

org.springframework.boot

spring-boot-starter

org.springframework.kafka

spring-kafka

org.springframework.boot

spring-boot-starter-test

test

org.springframework.kafka

spring-kafka-test

test

代码# application.yml

spring:

kafka:

bootstrap-servers: 127.0.0.1:9092

producer:

key-serializer: org.apache.kafka.common.serialization.StringSerializer

value-serializer: org.apache.kafka.common.serialization.StringSerializer

consumer:

key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

group-id: test-group// 启动类

@SpringBootApplication

public class DemoApplication {

public static void main(String[] args) {

SpringApplication.run(DemoApplication.class, args);

}

}

// 消费者

@Component

public class Consumer {

@KafkaListener(topics = { "test-topic" })

public void receiveMessage(String message) {

System.out.println(message);

}

}

// 生产者

@Component

public class Producer {

@Resource

KafkaTemplate kafkaTemplate;

public void sendMessage(String topic, String message) {

kafkaTemplate.send(topic, message);

}

}

// 测试

@RunWith(SpringRunner.class)

@SpringBootTest

public class DemoApplicationTests {

@Autowired

private Producer producer;

@Test

public void send() {

producer.sendMessage("test-topic", "test-message");

try {

Thread.sleep(10000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

总结

整合SpringBoot之后的代码还是非常简洁的,不过还是要熟悉原生API，这样才能在实际项目中遇到问题时游刃有余。