kafka安装及使用---Kafka从入门到精通（二）

.\bin\windows\kafka-server-start.bat .\config\server.properties
复制

 <!--引入kafak和spring整合的jar-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
复制

#------------------------------------Kafka-------------------#
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=localhost:9092
##########################producer about config##############################
spring.kafka.producer.acks=1
spring.kafka.producer.batch-size=16384
spring.kafka.producer.retries=0
spring.kafka.producer.buffer-memory=33554432
#spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer.class
#spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer.class
##########################consumer about config##############################
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.group-id=kafka_group_2
spring.kafka.consumer.auto-commit-interval=100
#spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer.class
#spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer.class
复制

/**
 * @author keying
 */
@RestController
public class TestKafkaController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 同步发送
     *
     * @return
     */
    @RequestMapping("/sendKafka")
    public String syncSendMessage() {
        for (int i = 0; i < 10; i++) {
            try {
                kafkaTemplate.send("kafka-boot", "0", "kafkaMessage" + i).get();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return "success";
    }


}



/**
 * @author keying
 */
@Component
public class ConsumerKafka {

    @KafkaListener(id = "foo", topics = "kafka-boot")
    public void listen1(String foo) {
        System.out.println("消费" + foo);
    }
}
复制