33 lines
983 B
Java
33 lines
983 B
Java
|
package com.cdzy.operations.component;
|
|||
|
|
|
|||
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|||
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|||
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|||
|
|
import lombok.extern.slf4j.Slf4j;
|
|||
|
|
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
|||
|
|
import org.springframework.kafka.annotation.KafkaListener;
|
|||
|
|
import org.springframework.stereotype.Component;
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* kafka消费者
|
|||
|
|
*
|
|||
|
|
* @author xiafan
|
|||
|
|
*/
|
|||
|
|
@Slf4j
|
|||
|
|
@Component
|
|||
|
|
public class KafkaConsumer {
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 消费者监听消息
|
|||
|
|
*
|
|||
|
|
* @param record 消息
|
|||
|
|
*/
|
|||
|
|
@KafkaListener(topics = {"msg_rsp"})
|
|||
|
|
public void onMessage(ConsumerRecord<?, ?> record) throws JsonProcessingException {
|
|||
|
|
log.info("[KAFKA接收] 主题: {}, 内容: {}", record.topic(), record.value());
|
|||
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|||
|
|
JsonNode jsonNode = objectMapper.readValue(record.value().toString(), JsonNode.class);
|
|||
|
|
//TODO：处理结果
|
|||
|
|
}
|
|||
|
|
}
|