特性
追求高吞吐量，适合产生大量数据的互联网服务的数据收集业务
kafka入门
1.导入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- kafkfa -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
</dependencies>
2.配置文件
生产者：
spring:
kafka:
bootstrap-servers: 192.168.200.130:9092 #主机IP:端口号
producer:
retries: 10
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
消费者
spring:
kafka:
bootstrap-servers: 192.168.200.130:9092 #主机IP:端口号
consumer:
group-id: ${spring.application.name}
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
3.生产者发送消息
@RestController
public class HelloController {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
//发送String
@GetMapping("/hello")
public String hello(){
String message = "Hello,SpringBoot Kafka!";
kafkaTemplate.send("topic",message);
return "success";
}
//发送pojo
@GetMapping("/sendPojo")
public String sendPojo(){
User user = User.builder()
.id(1001)
.name("张三")
.birthday(new Date())
.build();
kafkaTemplate.send("pojo-topic", JSON.toJSONString(user));
return "success";
}
}
3.消费者接收消息
@Component
public class HelloListener {
//接收Stirng
@KafkaListener(topics = "topic")
public void onMessage(String message){
if(!StringUtils.isEmpty(message)){
System.out.println(message);
}
}
//接收pojo
@KafkaListener(topics = "pojo-topic")
public void onMessage2(String message) {
if (!StringUtils.isEmpty(message)){
User user = JSON.parseObject(message, User.class);
System.out.println("user = " + user);
}
}
}
4.总结
- 生产者发送消息，多个消费者订阅同一个主题，只能有一个消费者收到消息（一对一）
- 生产者发送消息，多个消费者订阅同一个主题，所有消费者都能收到消息（一对多）（设置不同得组）
#kafka安装(docker)
Kafka对于zookeeper是强依赖，保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper
-
docker安装zookeeper
docker pull zookeeper:3.4.14创建容器
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14 -
Docker安装kafka
下载镜像：
docker pull wurstmeister/kafka:2.12-2.3.1创建容器
docker run -d --name kafka \ --env KAFKA_ADVERTISED_HOST_NAME=主机IP \ --env KAFKA_ZOOKEEPER_CONNECT=主机IP:2181 \ --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://主机IP:9092 \ --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \ --env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \ --restart always=true --net=host wurstmeister/kafka:2.12-2.3.1