SpringBoot配置Kafka

---

1. 设置kafka可以被外网访问（修改kafka下面的config下的server.properties）

//开启监听网址为服务器iplisteners=PLAINTEXT://192.168.111.100:9092//配置该kafka可以被访问的外网地址advertised.listeners=PLAINTEXT://ip:9092

配置后重启

1. SpringBoot配置kafka application.yml

spring:    kafka:    //刚才配置的kafka可以被外网访问的地址    bootstrap-servers: 192.168.111.100:9092    producer:      key-serializer: org.apache.kafka.common.serialization.StringSerializer      value-serializer: org.apache.kafka.common.serialization.StringSerializer    consumer:    //到kafka的config下consume中查看group-id      group-id: test-consumer-group      enable-auto-commit: true      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2. Producer

@Servicepublic class Producer {    private static final Logger logger = LoggerFactory.getLogger(Producer.class);   	//改Topic users需要到kfka下面新建话题，否则会报错    private static final String TOPIC = "users";    @Autowired    private KafkaTemplate kafkaTemplate;    public void sendMessage(String message) {        logger.info(String.format("生产者的消息为：%s", message));        //发送一条数据到kafka        this.kafkaTemplate.send(TOPIC, message);    }}

3. Consume

@Servicepublic class Consumer {    private final Logger logger= LoggerFactory.getLogger(Consumer.class);       @KafkaListener(topics = "users",groupId = "test-consumer-group")    public void consume(String message){        logger.info(String.format("消费者收到的消息为：%s",message));    }}

4. Controller

@RestController@RequestMapping("/kafka")public class KafkaController {    private Producer producer;    @Autowired    public KafkaController(Producer producer){        this.producer=  producer;    }    @RequestMapping("/publish")    public String sendMessageToKafkaTopic(@RequestParam("message") String message){        this.producer.sendMessage(message);        return "sucess";    }}