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