SpringBoot使用Kafka生产者、消费者

<!--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.JsonSerializer
    bootstrap-servers: 192.168.20.75:9907

kafka:
  spark:
    task:
      topic: platform-model-spark-topic1

方法一：添加@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();
    }

@KafkaListener(topics = "big_data_task_state", groupId = "bigDataTaskState")
    public void taskStateConsumer(String msg) {
        log.info("----receive:{}----", msg);
    }