前面说了kafka的topic有分区的概念，每个分区又有leader 和 follower，kafka听过ack机制保证消息的可靠性。
初识kafka---Kafka从入门到精通（一）
1、下载安装zookeeper
下载地址：http://zookeeper.apache.org/releases.html#download
1、进入解压地址，进入D:apache-zookeeper-3.8.0conf
2、 将“zoo_sample.cfg”重命名为“zoo.cfg”
3、 打开“zoo.cfg”找到并编辑dataDir=D:apache-zookeeper-3.8.0
4、添加系统变量：ZOOKEEPER_HOME=D:Kafkazookeeper-3.8.0
5、 编辑path系统变量，添加路径：%ZOOKEEPER_HOME%bin
6、 在zoo.cfg文件中修改默认的Zookeeper端口（默认端口2181）
7、 打开新的cmd，输入“zkServer“，运行Zookeeper
当出现下图，代表启动成功：
2、下载安装kafka
下载地址：https://kafka.apache.org/downloads
1、解压进入：D:kafka_2.13-3.1.0config
2、找到server.properties并打开，log.dirs=D:kafka_2.13-3.1.0kafka-logs
3、并编辑zookeeper.connect=localhost:2181
4、Kafka会按照默认，在9092端口上运行，并连接zookeeper的默认端口：2181
5、输入下面命令启动kafka，出现下图代表启动成功。
代码语言：javascript复制.binwindowskafka-server-start.bat .configserver.properties
3、集成springBoot
1、先pom文件加入kafka二方包
代码语言：javascript复制 <!--引入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);
}
}
