Spring Boot学习（三十三）：集成kafka

前言

下面是zookeeper和kafka的官网下载地址，大家可以学习下载

zookeeper下载地址：http://zookeeper.apache.org/releases.html

kafka下载地址：http://kafka.apache.org/downloads.html

1、添加依赖

在 pom.xml 文件中添加kafka依赖，依赖如下

	<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

2、配置Kafka信息

在 application.properties（或 application.yml）文件中配置 Kafka 的相关信息，下面是一个简单的示例：

#kafka地址，多个地址使用,分隔
spring.kafka.bootstrap-servers=127.0.0.1:9092
#消费者组ID
spring.kafka.consumer.group-id=myGroup
#序列化和反序列化
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3、发送消息

因为我们是springboot项目，已经集成了KafkaTemplate，我们可以直接使用KafkaTemplate来发送消息

下面，我编写一个发送消息的生产者

/*** 消息生产者*/
@Component
@Slf4j
public class KafkaProducer {@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate;/*** 发送消息* @param topic 主题* @param msg   消息*/public void send(String topic,String msg){kafkaTemplate.send(topic,msg).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onFailure(Throwable ex) {log.error("发送消息失败：{}", ex);}@Overridepublic void onSuccess(SendResult<String, String> result) {log.info("发送消息成功:{}");}});}/*** 发送消息* @param topic* @param msg*/public void send(String topic, Object msg) {send(topic, JSONObject.toJSONString(msg));}}

编写好生产者之后，我们就可以使用生产者发送消息，如下

	@Autowiredprivate KafkaProducer kafkaProducer;@GetMapping("send")public void sendMsg(){kafkaProducer.send("my-topic","hello world");}

如果想定制KafkaTemplate，那么可以在配置类进行配置，如下所示

@Configuration
public class KafakaConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;/*** 配置属性* @return*/@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}/*** 定制KafkaTemplate* @return*/@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());kafkaTemplate.setDefaultTopic("myGroup");return kafkaTemplate;}}

4、消费消息

使用 @KafkaListener 注解创建 Kafka 消费者，并监听指定的主题。接收到消息后，可以通过方法参数来接收消息：

@Slf4j
@Component
public class KafkaConsumer {/*** 消费my-topic主题的消息* @param message*/@KafkaListener(topics = "my-topic",groupId = "myGroup1")public void  receiveMessage(String message){log.info("消费消息："+message);}
}

同一消费者组，只会有一个消费者进行消费，如果想配置多个消费者同时处理，可以使用 @KafkaListener 注解来配置多个消费者。每个消费者需要配置不同的 group-id，监听主题一致，如下所示，就会有两个消费者同时消费

@Slf4j
@Component
public class KafkaConsumer {@KafkaListener(topics = "my-topic",groupId = "myGroup1")public void  receiveMessage(String message){log.info("消费消息："+message);}@KafkaListener(topics = "my-topic",groupId = "myGroup2")public void  receiveMessage2(String message){log.info("消费消息："+message);}}