springBoot对接kafka,批量、并发、异步获取消息,并动态、批量插入库表

kafka:    bootstrap-servers: 10.1.*.*:9092     #服务器的ip及端口，可以写多个，服务器之间用“：”间隔    producer: #生产者配置       key-serializer: org.apache.kafka.common.serialization.StringSerializer      value-serializer: org.apache.kafka.common.serialization.StringSerializer    consumer: #消费者配置      #指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名      group-id: myGroup                 #设置消费者的组id default：Group      enable-auto-commit: true  #设置自动提交offset      auto-commit-interval: 2000  #默认值为5000      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      #值的反序列化方式      value-serializer: org.apache.kafka.common.serialization.StringSerializer      auto-offset-reset: latest      max-poll-records: 2000  #批量一次最大拉取数据量 默认500    listener:      # poll-timeout: 1000      type: batch  # 开启批量消费      concurrency: 3  #指定listener 容器中的线程数，用于提高并发量    properties:      session:        timeout:          ms: 120000  #默认10000        max:          poll:            interval:              ms: 600000  #默认300000(5分钟)

spring:  redis:    # 地址    host: 127.0.0.1    # 端口，默认为6379    port: 6379    # 密码    # 连接超时时间    timeout: 10s    lettuce:      pool:        # 连接池中的最小空闲连接        min-idle: 0        # 连接池中的最大空闲连接        max-idle: 8        # 连接池的最大数据库连接数        max-active: 8        # #连接池最大阻塞等待时间（使用负值表示没有限制）        max-wait: -1ms

package com.wonders.config;
import com.fasterxml.jackson.annotation.JsonAutoDetect;import com.fasterxml.jackson.annotation.PropertyAccessor;import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.cache.annotation.CachingConfigurerSupport;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.data.redis.connection.RedisConnectionFactory;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;import org.springframework.data.redis.serializer.StringRedisSerializer;
/** * 〈自定义redis序列化方式〉 * @author yangyalin * @create 2018/11/1 * @since 1.0.0 */@Configurationpublic class RedisConfig extends CachingConfigurerSupport {    /**     * @Author yangyalin     * @Description redisTemplate序列化使用的jdkSerializeable, 存储二进制字节码（默认）, 所以自定义序列化类     * 用于存储可视化内容     * @Date 15:07 2018/11/1     * @Param [redisConnectionFactory]     * @return org.springframework.data.redis.core.RedisTemplate     **/    @Bean    public RedisTemplate        RedisTemplate        redisTemplate.setConnectionFactory(redisConnectionFactory);        //使用jackson2JsonRedisSerializer替换默认序列化        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer=new Jackson2JsonRedisSerializer(Object.class);        ObjectMapper objectMapper=new ObjectMapper();        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);        //设置key和value的序列化规则        redisTemplate.setKeySerializer(new StringRedisSerializer());        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);        redisTemplate.setHashKeySerializer(jackson2JsonRedisSerializer);        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);        redisTemplate.afterPropertiesSet();        return redisTemplate;    }}

        "insertMsgInfoTemp" parameterType="com.wonders.entity.KafkaMsgConfig">      INSERT INTO ${logTableName}("EVN_LOG_ID", "TABLE_NAME", "OPERATION", "PK_VALUE1", "PK_VALUE2",           "PK_VALUE3", "PK_VALUE4", "PK_VALUE5", "TRANS_FLAG", "PKS", "BASE_CODE", "PLA_BRANCH_CODE",           "CREATE_TIME","MSG_PRODUCE_TIME")      VALUES (#{id,jdbcType=VARCHAR}, #{tableName,jdbcType=VARCHAR}, #{operation,jdbcType=VARCHAR},            #{pk1,jdbcType=VARCHAR}, #{pk2,jdbcType=VARCHAR},#{pk3,jdbcType=VARCHAR},            #{pk4,jdbcType=VARCHAR},#{pk5,jdbcType=VARCHAR}, 'Y',            #{pks,jdbcType=VARCHAR}, #{baseCode,jdbcType=VARCHAR},            #{plaBranchCode,jdbcType=VARCHAR},sysdate,#{msgProduceTime,jdbcType=VARCHAR})    

    public void batchInsert(List kafkaMsgInfoList) throws Exception{        //如果自动提交设置为true,将无法控制提交的条数，改为最后统一提交        // 创建session实列        SqlSessionFactory sqlSessionFactory = ApplicationContextUtils.getBean("sqlSessionFactory");        // 开启批量处理模式 BATCH 、关闭自动提交事务 false        SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH,false);        KafkaMsgConfigMapper KafkaMsgMapper = sqlSession.getMapper(KafkaMsgConfigMapper.class);        int BATCH = 1000;        for (int i = 0,size=kafkaMsgInfoList.size(); i < size; i++) {            //循环插入 + 开启批处理模式            KafkaMsgMapper.insertKafkaMsgInfo(kafkaMsgInfoList.get(i));            if (i != 0 && i % BATCH == 0) {                sqlSession .commit();            }        }        // 一次性提交事务        sqlSession.commit();        // 关闭资源        sqlSession.close();    }

    @KafkaListener(topics = {"${mykafka.topics:mytopic}"})    public void myMQConsumer(List msgList){        log.info("接收到的消息条数size："+msgList.size());        //计算程序耗时时间        StopWatch stopWatch = new StopWatch();        // 开始计时        stopWatch.start();        this.getKafkaMsgAndDel(msgList);  //2、接收kafka日志并解析        stopWatch.stop();        log.info("本次任务耗时(秒)：" + stopWatch.getLastTaskTimeMillis()/1000 + "s");    }