Springboot项目中使用Kafka

叁歲伎倆 2023-03-06 05:12 14阅读 0赞

# Springboot项目中使用Kafka #

## 第一步：安装好Kafka服务器 ##

**具体可参考：** [https://blog.csdn.net/weixin\_40990818/article/details/107793976][https_blog.csdn.net_weixin_40990818_article_details_107793976]

## 第二步：创建Kafka生产者 ##

**a.引入依赖**

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

**b.spring-kafka-producer.xml：以XML的方式配置生产者**

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd">
    
        
        <bean id="producerProperties" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="127.0.0.1:9092"/>
                    
                    <entry key="acks" value="1"/>
                    
                    <entry key="retries" value="1"/>
                    
                    <entry key="request.timeout.ms" value="2000"/>
                    
                    <entry key="max.block.ms" value="2000"/>
                    <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                    <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                </map>
            </constructor-arg>
        </bean>
    
    
        
        <bean id="defaultKafkaProducerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
            <constructor-arg ref="producerProperties"/>
        </bean>
    
    
        
        <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
            <constructor-arg ref="defaultKafkaProducerFactory"/>
            
            <constructor-arg name="autoFlush" value="false"/>
        </bean>
    </beans>

**c.在启动类引入spring-kafka-producer.xml配置类**

@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class})
    @ImportResource("classpath:*spring-kafka-producer.xml")
    public class UsermanageAPP extends SpringBootServletInitializer { 
        private static Logger logger= LogManager.getLogger(LogManager.ROOT_LOGGER_NAME);
    
        @Override
        protected SpringApplicationBuilder configure(SpringApplicationBuilder application) { 
            return application.sources(UsermanageAPP.class);
        }
    
        /** * 项目的启动方法 * @param args */
        public static void main(String[] args) { 
            SpringApplication.run(UsermanageAPP.class, args);
            logger.info("======服务已经启动========");
        }
    }

**d.使用kafkaTemplate发送消息**

@RestController
    public class UserManageController { 
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        public static final String TOPIC_NAME = "my-replicated-topic";
    
        @RequestMapping("/hello")
        public void test(){ 
            kafkaTemplate.send(TOPIC_NAME, "123");
        }
    
    }

**e.访问127.0.0.1:8080/hello后，进入kafka中进行验证**  
![在这里插入图片描述][2020080514180880.png]

## 第二步：创建Kafka消费者 ##

**a.引入依赖**

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

**b.spring-kafka-consumer.xml：以XML的方式配置消费者**  
groupId：consumer group是kafka提供的可扩展且具有容错性的消费者机制。一个组可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。  
组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。当然，每个分区只能由同一个消费组内的一个consumer来消费，从而保证消息不会被重复消费。

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:context="http://www.springframework.org/schema/context"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd
           http://www.springframework.org/schema/context
           http://www.springframework.org/schema/context/spring-context.xsd">
    
    
    
        <bean id="consumerProperties" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="127.0.0.1:9092" />
                    
                    <entry key="enable.auto.commit" value="false" />
                    <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" />
                    <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" />
                </map>
            </constructor-arg>
        </bean>
    
        
        <bean id="consumerFactory"
              class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
            <constructor-arg ref="consumerProperties"/>
        </bean>
    
    
        
        
        
        <bean id="tradeprodMessageListener" class="com.luo.nacos.kafka.TradeprodMessageListener" />
        
        <bean id="tradeprodProperties" class="org.springframework.kafka.listener.ContainerProperties">
            <constructor-arg value="my-replicated-topic" />
            
            <property name="ackMode" value="RECORD" />
            <property name="groupId" value="my-replicated-topic" />
            
            <property name="messageListener" ref="tradeprodMessageListener" />
        </bean>
        
        <bean id="tradeprodListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer"
              init-method="doStart">
            <constructor-arg ref="consumerFactory" />
            <constructor-arg ref="tradeprodProperties" />
            
            <property name="concurrency" value="3" />
        </bean>
        
    
    
    </beans>

**c.启动类引入spring-kafka-consumer.xml配置文件**

@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class})
    @ImportResource("classpath:*spring-kafka-consumer.xml")
    public class NacosApp extends SpringBootServletInitializer { 
        private static Logger logger= LogManager.getLogger(LogManager.ROOT_LOGGER_NAME);
    
        @Override
        protected SpringApplicationBuilder configure(SpringApplicationBuilder application) { 
            return application.sources(NacosApp.class);
        }
        /** * 项目的启动方法 * @param args */
        public static void main(String[] args) { 
            SpringApplication.run(NacosApp.class, args);
            logger.info("======服务已经启动========");
        }
    }

**d.TradeprodMessageListener ：消息消费的类要实现MessageListener接口，重写onMessage方法**

public class TradeprodMessageListener implements MessageListener<String, String> { 
    
        @Override
        public void onMessage(ConsumerRecord<String, String> record) { 
            try { 
                if (record != null && record.value() != null) { 
                    System.out.println("收到kafka消息:"+ record.value());
                }
            } catch (Exception e) { 
                System.out.println("消费kafka消息失败，失败原因:"+ e.getMessage());
            }
        }
    }

**e.利用生产者产生消息进行验证**  
![在这里插入图片描述][20200805141744350.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]

[https_blog.csdn.net_weixin_40990818_article_details_107793976]: https://blog.csdn.net/weixin_40990818/article/details/107793976
[2020080514180880.png]: /images/20230208/e522f7aa004e4e66b1451f2978b2475b.png
[20200805141744350.png]: /images/20230208/a14b3fd4f6974418b35c8049bcd635c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]: /images/20230208/d7155a1be72e4b6caae629863f7a2376.png