[大数据]kafka学习记录

kafka学习记录
1.生产者

//添加依赖
    @Autowired
    KafkaTemplate kafkaTemplate;
// 调用
	public boolean send(Map<String,Object> data){
		try {
			//自定义topic
			String topic = "kafka-topic";
			//将数据转为json  此处用的是hutool的JSONUtil   data是可以经过二次处理后再发送到kafka  此处省略
			String message = JSONUtil.toJsonStr(data);
	        if (log.isInfoEnabled()) {
                log.info("向{}发送统计消息：{}", topic, message);
            }
            kafkaTemplate.send(topic, message);
            return true;
		 } catch (Exception e) {
                String errorInfo = String.format("数据统计发送Kafka异常:%s\n%s", e.getMessage(), JSONUtil.toJsonStr(vo));
                //@Slf4j 
                log.error(errorInfo);
                return false;
            }
	}

2.消费者
监听topic

@Component
public class Consumer(){
    @KafkaListener(topics = { "kafka-topic" })
    public void listen(ConsumerRecord<?, ?> record) {
    //可以二次处理   
    //如果不是需要处理的信息  可以通过logstash 上传到es  不需要写消费者方法  直接es监听topic
    //需要二次处理转存到mysql、redis之类的   需要写此方法监听
     String recordStr = record.value().toString();
     JSONObject jsonObject = JSONUtil.parseObj(recordStr);
     ...
    }
}