@
目录
- 依赖
- 配置文件
- 生产者
- 消费者
依赖
<!--kafka-->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>xxx</version>
</dependency>
配置文件
spring:kafka:producer:value-serializer: org.springframework.kafka.support.serializer.JsonSerializerbootstrap-servers: 192.168.20.75:9907kafka:spark:task:topic: platform-model-spark-topic1生产者
方法一：添加@RunWith(SpringRunner.class)、@SpringBootTest(classes = DataComputingModelApplication.class)实现初始化配置注入kafkaTemplate，调用send()
@Autowired
private KafkaTemplate kafkaTemplate;@Testpublic void kafkaSend() {final ProducerRecord<String, String> record = new ProducerRecord("test20201228", "{\"key\":\"27\"}");kafkaTemplate.send(record);log.info("------------send success!----------------");}方法二：不需要注解@RunWith、@SpringBootTest，但是初始化Properties，同样调用send()
@Testpublic 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();}
消费者
说明：
① 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);}
重要信息
- 官网：https://ais.cn/u/vEbMBz
