@[TOC]
依赖
<!--kafka--> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>xxx</version> </dependency>AI 代码解读
配置文件
spring:
kafka:
producer:
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
bootstrap-servers: 192.168.20.75:9907
kafka:
spark:
task:
topic: platform-model-spark-topic1
AI 代码解读
生产者
方法一：添加@RunWith(SpringRunner.class)、@SpringBootTest(classes = DataComputingModelApplication.class)实现初始化配置注入kafkaTemplate，调用send()
@Autowired
private KafkaTemplate kafkaTemplate;
@Test
public void kafkaSend() {
final ProducerRecord<String, String> record = new ProducerRecord("test20201228", "{\"key\":\"27\"}");
kafkaTemplate.send(record);
log.info("------------send success!----------------");
}
方法二：不需要注解@RunWith、@SpringBootTest，但是初始化Properties，同样调用send()
@Test
public void kafkaSend2() {
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.20.75:9907");
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<String, String>("test20201228", "key", "{\"key\":\"20\"}");
producer.send(record);
log.info("------------send success!----------------");
producer.close();
}
AI 代码解读
消费者
==说明：
① Topic主题用来区分不同类型的消息
② GroupId用来解决同一个Topic主题下重复消费问题，比如一条消费需要多个消费者接收到，就可以通过设置不同的GroupId实现，实际消息是存一份的，只是通过逻辑上设置标识来区分，系统会记录Topic主题下--》GroupId分组下--》partition分区下的offsert，来标识是否消费过。==
@KafkaListener(topics = "big_data_task_state", groupId = "bigDataTaskState")
public void taskStateConsumer(String msg) {
log.info("----receive:{}----", msg);
}
AI 代码解读
重要信息
