SpringMVC整合Kafka实战

电玩女神 2023-06-25 03:12 18阅读 0赞

# 1.SpringMVC整合生产者(Producer) #

SpringMVC整合生产者比较简单，我直接用一个单例对象来管理生产者，这样保证生产者也是一个单例对象。

## 1.1 导入kafka的maven依赖 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70][]

## 1.2 创建单例对象管理生产者 ##

我使用静态内部类的方式创建单例对象，保证单例对象的线程安全。直接上代码

public class KafkaManager {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(KafkaManager.class);
    
        private static final String BOOTSTRAP_SERVERS = Config.getMqConfig("kafka.bootstrap.servers");
    
    
        public static final String TOPIC_LOGIN = "Topic_login";
        public static final String TOPIC_OLD_API = "Topic_api_use";
        public static final String TOPIC_COURSE_VISIT = "Topic_course_visit";
    
        private boolean kafkaAvailable = true;
    
        private KafkaProducer<String, String> producer;
    
        private KafkaManager() {
            this.initManger();
        }
        private static class SingletonManager {
            private static final KafkaManager singletonObj = new KafkaManager();
        }
    
        public static KafkaManager getInstance() {
            return KafkaManager.SingletonManager.singletonObj;
        }
    
        private void initManger() {
            try {
                Properties producerProperties = new Properties();
                //设置接入点，请通过控制台获取对应 Topic 的接入点
                producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
                //消息队列 Kafka 消息的序列化方式
                producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                //请求的最长等待时间
                producerProperties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
                this.producer = new KafkaProducer<String, String>(producerProperties);
    
            } catch (Exception ex) {
                LOGGER.error("创建kafka失败",ex);
                this.kafkaAvailable = false;
            }
        }
    
        protected void finalize () {
            if (null != this.producer) {
                this.producer.close();
            }
        }
    
        public boolean sendMessage(String topic, JSONObject body) {
            if (false == this.kafkaAvailable) {
                return false;
            }
            boolean succeed = true;
            try {
                ProducerRecord<String, String>  kafkaMessage =  new ProducerRecord<String, String>(topic, body.toJSONString());
    
                //发送消息，并获得一个 Future 对象
                Future<RecordMetadata> metadataFuture = this.producer.send(kafkaMessage);
                
            } catch (Exception e) {
                LOGGER.error("发送kafka失败",e);
                succeed = false;
            }
            return succeed;
        }
    
    }

## 1.3 生产者发送消息 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 1][]

# 2.SpringMVC整合消费者（Consumer） #

相对于生产者，SpringMVC整合消费者就显得复杂一点。

## 2.1 引入kafka消费者的maven依赖 ##

<dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka-clients</artifactId>
          <version>0.10.0.0</version>
        </dependency>
    
        <dependency>
          <groupId>org.springframework.kafka</groupId>
          <artifactId>spring-kafka</artifactId>
          <version>1.1.1.RELEASE</version>
        </dependency>

如上所示，除了kafka客户端的依赖外还需要spring-kafka的依赖。

这里说明一下，我使用Spring配置文件创建bean的方式来管理kafka的消费者。当然还有其他方式，这里我选了比较简单的一种，其他方式可以自行百度一下。

## 2.2 创建监听类 ##

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.listener.MessageListener;
    
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    
    public class TopicApiUseKafkaListener implements MessageListener<String, String> {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(TopicApiUseKafkaListener.class);
    
        private static final String TABLE_PRE = "";
    
        @Override
        public void onMessage(ConsumerRecord<String, String> stringStringConsumerRecord) {
            IApiUseService apiUseService = SpringContextUtil.getBean("apiUseService");
            List<ApiUse> dataList = new ArrayList<>();
            String tableName = TABLE_PRE + DateUtil.formatDate(new Date(),"yyyyMM");
            String value = stringStringConsumerRecord.value();
            JSONObject bodyJson = JSONObject.parseObject(value);
    
            Integer platform = bodyJson.getInteger("platform");
            String url = bodyJson.getString("url");
            String apiVersion = bodyJson.getString("apiVersion");
            String createDate = bodyJson.getString("createDate");
    
            if (!StringUtils.isEmpty(url) && null != platform && !StringUtils.isEmpty(createDate) && !StringUtils.isEmpty(apiVersion)) {
                ApiUse item = new ApiUse();
                item.setPlatform(platform);
                item.setUrl(url);
                item.setApiVersion(apiVersion);
                item.setCreateDate(createDate);
                dataList.add(item);
    
            }
            if (dataList.size() > 0) {
                try {
                    apiUseService.addApiUse(tableName, dataList);
                } catch (Exception ex1) {
                    LOGGER.error("api使用插入错误",ex1);
                }
            }
        }
    }

## 2.3 配置监听类 ##

kafka的监听类的配置需要配置在dispatcher-servlet.xml文件中，配置在spring-context.xml文件中是无效的，这是由Spring的加载机制决定的。

先定义consumer的参数再创建一个全局的consumerFactory的bean对象，如下图所示。这里需要注意enable.auto.commit这个参数，这是kafka自动提交的参数，如果设置成true就有kafka服务端来自动提交，但是由于网络等因素kafka服务端自动提交有时候会失败导致重复消费消息。所以这里设置成false，虽然设置成false但是实际上不需要开发者来手动提交，提交的操作是由Spring-kafka这个sdk类处理的，有兴趣的童鞋可以看一下spring-kafka的源码就知道了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 2][]

接下来创建实际执行消费的类，就是我们刚才创建的那个类。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 3][]

如上图所示，原则上每个topic就要创建一个对应的监听类，然后再配置监听类需要监听的topic。

内容就到这里。更多精彩文章敬请期待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191225105533899.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191225105812390.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/201912251106221.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2019122511115877.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpeXV4NDg2OQ==,size_16,color_FFFFFF,t_70