Spring Boot操作Kafka
(1).Maven的配置
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.10.2.0</version></dependency>
(2).增加配置文件: application.properties
# kafka连贯地址,多个地址之间应用","隔开spring.kafka.bootstrap-servers = master:9092# 生产者配置spring.kafka.producer.acks = 0spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.retries = 3spring.kafka.producer.batch-size = 4096spring.kafka.producer.buffer-memory = 33554432spring.kafka.producer.compression-type = gzip# 消费者配置spring.kafka.consumer.group-id = mygroupspring.kafka.consumer.auto-commit-interval = 5000spring.kafka.consumer.heartbeat-interval = 3000spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.StringDeserializerspring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.StringDeserializerspring.kafka.consumer.auto-offset-reset = earliestspring.kafka.consumer.enable-auto-commit = true# listenner, 标识消费者监听的个数spring.kafka.listener.concurrency = 8# topic的名字kafka.topic1 = topic1
(3).生产者代码
import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Value;import org.springframework.kafka.core.KafkaTemplate;@Service@Slf4jpublic class MyKafkaProducerServiceImpl implements MyKafkaProducerService { @Resource private KafkaTemplate<String, String> kafkaTemplate; // 读取配置文件 @Value("${kafka.topic1}") private String topic; @Override public void sendKafka() { kafkaTemplate.send(topic, "hell world"); }}
(4),消费者代码
@Component@Slf4jpublic class MyKafkaConsumer { @KafkaListener(topics = "${kafka.topic1}") public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { log.info("----------------- record =" + record); log.info("------------------ message =" + kafkaMessage.get());}