kafka 消息模式（kafka-消息中间键(springboot集成)）

特性

追求高吞吐量   
    
    
    
，适合产生大量数据的互联网服务的数据收集业务

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