Kafka学习进阶(五)--记一次线上调优(max.request.size)和kafka线上解决思路

1、回顾kafka的知识点和问题陈述

        前面几篇kafka文章，介绍从搭建到优化。   

Kafka消息队列学习(一)--场景和搭建;

Kafka消息队列学习进阶(二)--项目实战;

Kafka消息队列学习进阶(三)--原理概念介绍篇;

Kafka消息队列学习进阶(四)--优化(配置/代码/集群);

        以上都是在test和uat环境进行测试和迁移数据的，最近迁移项目正式上线，可是上线当晚就出现严重的bug，现象是：

        1.一执行迁移程序，数据库链接就超时。

       2.kafka链接不上，同时查看error.log日志，kafka一直报数据大小超过kafka最大的发送size(kafka max.request.size)。也就是那天根本没有上线成功是失败的。那么咱们是怎么处理的呢？

 

2、解决问题的思路--优化配置

        首先解决的是数据库链接不上的问题，一是检查数据库是否启动成功，检查完毕，启动成功。是启动程序致使的，并非一开始就链接不上。继续排查，看info.log日志，看看程序执行到哪里，判断哪里在调用数据库，最后根据sql反查生产库，发现是这条语句查询的数据有35W，可是数据库根本承受不住那么大返回数据。最后更改查询区间大小，胜利解决这个问题。

      如今才开始来解决kafka的问题，因为本人，对kafka配置信息的不熟悉，或者说是生产库和uat库/test库的数据根本不是一个档次的状况不清楚，没有预估到致使的。建议对数据迁移的项目，最好想预估其有多大的数据，而后采用不一样的方式。

        因为第一步已解决数据库的链接问题，已经减少了数据库大小，同时返回的数据也减小了。kafka已经在发送数据，日志一直在打印"数据发送成功！"，但是没想到的是，kafka根本没有消费，由于消费日志没有打印啊。info.log日志打印的是发送成功，可是数据又没有写入到数据库，很奇怪啊，当时心情真的是.....而后又去查看error.log日志，发现一直报：kafka max.request.size的问题，就是发送的数据大小已经超过kafka的配置发送数据大小，致使一直发送不过去。因为没有配置过，只能百度。下面以yml配置为例：

消费数据：

    properties:

        max.partition.fetch.bytes: 15000000

生产数据：

    properties:

        max.request.size: 15000000

        以上，算是从新配置了，OK，从新启动服务器，再来一次迁移程序的执行。但是新的问题又出现了，生产数据(发送)和消费数据没有问题，可是很慢，没有测试那么快。同时，越日后执行，越慢，最终线程池报错，超过线程大小(解释一下：发送数据是异步的)。因此说优化配置参数是失败的。那只能另外想办法了。

代码优化

        根据问题，我想到的是，以前在插入数据库的时候，若是数据过于大，也会致使数据库插入失败或者说很慢，是分批次提交插入数据到数据库的。因此最终咱们的想法也是：查询出来的数据，分批次发送到kafka，顺利解决消费慢和发送数据过于大的问题。具体代码以下：

 executorService.submit(() -> {

            //查询会员的数据

            List<Class> dataSize = mapper.get();

            if (StringUtils.isObjNotEmpty(dataSize) && dataSize.size() > 0) {

                //限制条数

                int pointsDataLimit = sendSize;

                Integer size = dataSize.size();

                //判断是否有必要分批

                if (pointsDataLimit < size) {

                    //分批数

                    int part = size / pointsDataLimit;

                    for (int i = 0; i < part; i++) {

                        //10000条

                        List<Class> listPage = dataSize.subList(0, pointsDataLimit);

                        //发送数据

                        sendMessage(integer,sizePage,time,endTime,accountType,listPage);

                        //剔除

                        dataSize.subList(0, pointsDataLimit).clear();

                    }

                    if (!dataSize.isEmpty()) {

                        sendMessage(integer,sizePage,time,endTime,accountType,dataSize);

                    }



                } else {

                    sendMessage(integer,sizePage,time,endTime,accountType,dataSize);

                }



            }else{

              

                log.info("数据为空的数据段：{}",integer+":"+sizePage);

            }

   });

   

private void sendMessage(Integer integer, Integer sizePage, String time, String endTime, String accountType, List<Class> listPage) {

        //查询

        Message message = new Message();

        message.setId(String.valueOf(integer+":"+sizePage+":"+time+":"+endTime+":"+accountType));

        message.setMsg(JSON.toJSONString(listPage));

        message.setSendTime(new Date());

        log.info("数据发送成功的数据段：{}",integer+":"+sizePage);

        try {

            kafkaTemplate.send("userAndAccount", JSONUtil.toJsonStr(message));

        }catch (Exception e){

            redisTemplate.opsForList().leftPush(RedisKeyConstsnts.RECORD_ERROR_KEY,"数据迁移迁移数据失败,发送消息失败，该批次信息:"+integer+":"+endTime);

            log.info("数据迁移迁移数据失败,发送消息失败，该批次信息，data：{}",integer+":"+endTime);

            e.printStackTrace();

        }

    }

3、总结

整理问题：

        1.熟悉线上环境数据有多大。

        2.认真熟悉kafka的配置文件。同时配置越大并不必定是万能的，须要配置加上代码相互。

补充知识点：

Kafka设计的初衷是迅速处理短小的消息，通常10K大小的消息吞吐性能最好（可参见LinkedIn的kafka性能测试）。但有时候，咱们须要处理更大的消息，好比XML文档或JSON内容，一个消息差很少有10-100M，这种状况下，Kakfa应该如何处理?
针对这个问题，有如下几个建议：

最好的方法是不直接传送这些大的数据。若是有共享存储，如NAS, HDFS, S3等，能够把这些大的文件存放到共享存储，而后使用Kafka来传送文件的位置信息。

第二个方法是，将大的消息数据切片或切块，在生产端将数据切片为10K大小，使用分区主键确保一个大消息的全部部分会被发送到同一个kafka分区（这样每一部分的拆分顺序得以保留），如此以来，当消费端使用时会将这些部分从新还原为原始的消息。

第三，Kafka的生产端能够压缩消息，若是原始消息是XML，当经过压缩以后，消息可能会变得不那么大。在生产端的配置参数中使用compression.codec和commpressed.topics能够开启压缩功能，压缩算法可使用GZip或Snappy。

不过若是上述方法都不是你须要的，而你最终仍是但愿传送大的消息，那么，则能够在kafka中设置下面一些参数：
broker 配置:

     ​    ​message.max.bytes (默认:1000000) – broker能接收消息的最大字节数，这个值应该比消费端的fetch.message.max.bytes更小才对，不然broker就会由于消费端没法使用这个消息而挂起。

    ​    ​log.segment.bytes (默认: 1GB) – kafka数据文件的大小，确保这个数值大于一个消息的长度。通常说来使用默认值便可（通常一个消息很难大于1G，由于这是一个消息系统，而不是文件系统）。

replica.fetch.max.bytes (默认: 1MB) – broker可复制的消息的最大字节数。这个值应该比message.max.bytes大，不然broker会接收此消息，但没法将此消息复制出去，从而形成数据丢失。

consumer 配置:

     ​    ​fetch.message.max.bytes (默认 1MB) – 消费者能读取的最大消息。这个值应该大于或等于message.max.bytes。

 因此，若是你必定要选择kafka来传送大的消息，还有些事项须要考虑。要传送大的消息，不是当出现问题以后再来考虑如何解决，而是在一开始设计的时候，就要考虑到大消息对集群和主题的影响。

性能: 根据前面提到的性能测试，kafka在消息为10K时吞吐量达到最大，更大的消息会下降吞吐量，在设计集群的容量时，尤为要考虑这点。

可用的内存和分区数：Brokers会为每一个分区分配replica.fetch.max.bytes参数指定的内存空间，假设replica.fetch.max.bytes=1M，且有1000个分区，则须要差很少1G的内存，确保 分区数*最大的消息不会超过服务器的内存，不然会报OOM错误。一样地，消费端的fetch.message.max.bytes指定了最大消息须要的内存空间，一样，分区数*最大须要内存空间 不能超过服务器的内存。因此，若是你有大的消息要传送，则在内存必定的状况下，只能使用较少的分区数或者使用更大内存的服务器。

垃圾回收：到如今为止，我在kafka的使用中还没发现过此问题，但这应该是一个须要考虑的潜在问题。更大的消息会让GC的时间更长（由于broker须要分配更大的块），随时关注GC的日志和服务器的日志信息。若是长时间的GC致使kafka丢失了zookeeper的会话，则须要配置zookeeper.session.timeout.ms参数为更大的超时时间。

一切的一切，都须要在权衡利弊以后，再决定选用哪一个最合适的方案。

繁荣Aaron

没时间解释了，快长按左边二维码关注咱们～～