springboot simple （8） springboot kafka

﹏ヽ暗。殇╰゛Y 2024-04-01 16:07 22阅读 0赞

## 1 kafka简介 ##

Apache Kafka是分布式发布 - 订阅消息系统。  
官网：[http://kafka.apache.org][http_kafka.apache.org]  
下载地址：[https://archive.apache.org/dist/kafka][https_archive.apache.org_dist_kafka]

Kafka 的架构包括以下组件：  
1、生产者（Producer）：发布消息；  
2、服务代理（Broker）：已发布的消息保存在一组服务器中，它们被称为代理（Broker）或 Kafka 集群；  
3、消费者（Consumer）：可以订阅一个或多个话题，并从 Broker接收的消息。  
![在这里插入图片描述][a6d3b18d33064e8db50ed37896120a71.png]

## 2 springboot集成kafka ##

### 2.1 建立生产者producer的工程； ###

第1步：POM文件引入：

<dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    			<version>2.9.2</version>
    		</dependency>

第2步：设置配置参数，修改application.properties文件：

# kafka连接接地址
    spring.kafka.bootstrap-servers= localhost:9092
    # 生产者配置
    # 序列化key的类
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer 
    # 反序列化value的类
    spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

第3步：定义ProducerService类，通过调用KafkaTemplate类来发送消息；

@Service
    public class ProducerService {
        
        public static final String topic = "test";
    
        @Autowired
        private KafkaTemplate<?,String> kafkaTemplate;
    
        //向topic发送消息
        public void sendMessage(String message){
        
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("data",message);
            kafkaTemplate.send(topic,jsonObject.toJSONString());
        }
    }

第4步：定义KafkaController类，通过http接口可以发送消息。

@RequestMapping(value = "/sendMessage", method = RequestMethod.GET)
        public String sendMessage(@RequestParam(value = "message") String message) {
        
            producerService.sendMessage(message);
            return  message;
        }

### 2.2 建立生产者consumer的工程 ###

第1步：POM文件引入：

<dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    			<version>2.9.2</version>
    		</dependency>

第2步：设置配置参数，修改application.properties文件：

# kafka连接接地址
    spring.kafka.bootstrap-servers=localhost:9092
    # 消费者所属消息组
    spring.kafka.consumer.group-id=testGroup 
    # 反序列化key的类
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer 
    # 反序列化value的类
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

第3步：定义通过调用ConsumerService类，通过@KafkaListener来实现消息的监听。

@Service
    @Slf4j
    public class ConsumerService {
        
        public static final String topic = "test";
    
        @Autowired
        private KafkaTemplate<?,String> kafkaTemplate;
    
        //从topic接收消息
        @KafkaListener(topics = "test", groupId = "testGroup", topicPartitions = {
        })
        public void receiveMessage(String message) {
        
            log.info("receive:" + message);
        }
    }

### 2.3 生产者（producer）和消费者（consumer）通信 ###

验证如下：Postman发送http get请求：  
![在这里插入图片描述][3946f012a3c8496d8908b8b940f4ffa8.png]

生产者consumer的工程控制输入log：  
![在这里插入图片描述][4eed338661294928a86d50e907eb7271.png]

可见：kafka服务启动成功，生产者（producer）消费者（consumer）通信正常。

代码详见：  
[https://gitee.com/linghufeixia/springboot-simple][https_gitee.com_linghufeixia_springboot-simple]  
chapter6-2： 生产者（producer）项目  
chapter6-3： 消费者（consumer）项目

--------------------

**教程列表**：  
[springboot simple（0） springboot简介][springboot simple_0_ springboot]  
[springboot simple（1） springboot Helloworld][springboot simple_1_ springboot Helloworld]  
[springboot simple（2） springboot Starter][springboot simple_2_ springboot Starter]  
[springboot simple（3 ）springboot Web开发][springboot simple_3 _springboot Web]  
[springboot simple（4）springboot 数据持久化][springboot simple_4_springboot]  
[springboot simple （5） springboot Nosql][springboot simple _5_ springboot Nosql]  
[springboot simple （6） springboot mqtt][springboot simple _6_ springboot mqtt]  
[springboot simple （7） springboot thrift][springboot simple _7_ springboot thrift]  
[springboot simple （8） springboot kafka][springboot simple _8_ springboot kafka]  
[springboot simple （9） springboot jpa（Hibernate）][springboot simple _9_ springboot jpa_Hibernate]  
[springboot simple （10） springboot protobuf][springboot simple _10_ springboot protobuf]  
[springboot simple （11） springboot protostuff][springboot simple _11_ springboot protostuff]  
[springboot simple （12） springboot RabbitMQ][springboot simple _12_ springboot RabbitMQ]

[http_kafka.apache.org]: http://kafka.apache.org
[https_archive.apache.org_dist_kafka]: https://archive.apache.org/dist/kafka
[a6d3b18d33064e8db50ed37896120a71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/98a8f7038a3a41d08736aebae2b891d5.png
[3946f012a3c8496d8908b8b940f4ffa8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/543faa14183b4534a123379536fdd2e3.png
[4eed338661294928a86d50e907eb7271.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/8c2a3342b2034877a94b7136a9949d27.png
[https_gitee.com_linghufeixia_springboot-simple]: https://gitee.com/linghufeixia/springboot-simple
[springboot simple_0_ springboot]: https://blog.csdn.net/afei8080/article/details/101297868
[springboot simple_1_ springboot Helloworld]: https://blog.csdn.net/afei8080/article/details/101345712
[springboot simple_2_ springboot Starter]: https://blog.csdn.net/afei8080/article/details/101346285
[springboot simple_3 _springboot Web]: https://blog.csdn.net/afei8080/article/details/101348597
[springboot simple_4_springboot]: https://blog.csdn.net/afei8080/article/details/101422030
[springboot simple _5_ springboot Nosql]: https://blog.csdn.net/afei8080/article/details/101422594
[springboot simple _6_ springboot mqtt]: https://blog.csdn.net/afei8080/article/details/121513387?spm=1001.2014.3001.5502
[springboot simple _7_ springboot thrift]: https://blog.csdn.net/afei8080/article/details/127687148
[springboot simple _8_ springboot kafka]: https://blog.csdn.net/afei8080/article/details/127749475
[springboot simple _9_ springboot jpa_Hibernate]: https://blog.csdn.net/afei8080/article/details/128001389
[springboot simple _10_ springboot protobuf]: https://blog.csdn.net/afei8080/article/details/128297408
[springboot simple _11_ springboot protostuff]: https://blog.csdn.net/afei8080/article/details/128342603
[springboot simple _12_ springboot RabbitMQ]: https://blog.csdn.net/afei8080/article/details/128493057