曾经参与的一次并发优化回顾

2024-06-12 23:32

本文主要是介绍曾经参与的一次并发优化回顾,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

业务场景:

我们有一个在线考试的模块,当时需要支持用户同时1000人的提交考试试卷。

问题:

测试者采用jmeter测试提交考试答案的接口,100个线程并发的时候。报错Max number of active transactions reached:50。错误率有25%。

问题分析:

我们在提交答案的接口采用了spring的事务@Transactional。我们采用的atomikos来实现分布式事务。在jta.properties中有com.atomikos.icatch.max_actives = 50。如果我们当前我们同时的事务数量大于配置的50的时候,就会报错。

尝试解决:

我们可以增加max_actives,但是我们的并发量大的时候,依然会有问题。

当很多用户同时提交答案的时候,我们同时会有大量的事务处理,对我们的数据库mysql的压力也比较大。我们可以采用队列的形式来削峰,我们用这种异步的形式把请求的数据放到队列,给前端提示一个考试结果正在计算中的展示,我们再从队列中拿数据去消费。我们用的是kafka

实现:

直接采用kafkaTemplate 发送数据,再设置一个监听KafkaListener,获取对应的topic数据

进一步问题:

改进之后,可以支持一定并发,但是速度并不快。

再次分析:

通过观察服务器性能,发现这个问题出现在io上,我们提交的对象包含所有题目的答案,我们的题目数量很多,100道题目时候,还有可能存在简答题的时候,我们这个传过去的对象就很大了,一个空的String对象有24字节,一个题目包含一个Integer的试题类型、Object的考生答案、Long类型的试卷Id。因为是包装类型,含有对象头信息。64位没有开启压缩指针Interger=Header(标记头8字节 + 对象指针8字节) + int(4字节)+ 对齐(4) = 24字节。可以看出一个题目大约有100个字节,100个题目加上用户的基础信息,差不多1万个字节,也差不多是10KB,1000人同时提交就有10M。

我们可以把这个对象优化,序列化我们自己的可以解析,更小的对象传给kafka。我们在kafka中一般会配置producer和consumer的key、value的serializer,一般都是配置StringSerializer,kafka默认提供了一些(de)Serializer,String、byteBuffer、ByteArray、Bytes、Long等,这里我们采用ByteBuffer,可以把我们的数据压缩到很小,这样我们传给kafka的数据就会小很多。

代码

// 序列化我们的对象
public ByteBuffer serialize(TestAnswerVO vo) {
    ByteBuffer byteBuffer = ByteBuffer.allocate(calculateCapacity(vo));
    byteBuffer.putLong(obtainType(vo.getUserId()));
    byteBuffer.putLong(obtainType(vo.getTestId()));
    byteBuffer.putInt(vo.getCostTime());

    byte[] accounts = vo.getAccount().getBytes();
    byteBuffer.putInt(accounts.length);
    if (accounts.length > 0)
        byteBuffer.put(vo.getAccount().getBytes());

    if (CollectionUtils.isNotEmpty(vo.getList())) {
        byteBuffer.putInt(vo.getList().size());
        for (ExaAnswerCardVO e : vo.getList()) {
            byteBuffer.putLong(obtainType(e.getTestPageId()));
            byteBuffer.putLong(obtainType(e.getExaResultId()));
            byteBuffer.putInt(e.getType());

            if (Objects.isNull(e.getExaAnswer())) {
                byteBuffer.putInt(0);
            } else {
                byte[] bytes = e.getExaAnswer().toString().getBytes();
                byteBuffer.putInt(bytes.length);
                if (bytes.length > 0) {
                    byteBuffer.put(bytes);
                }
            }
        }
    }
    byteBuffer.flip();
    return byteBuffer;
}


// 反序列化对象 我们只需要从ByteBuffer按照我们put的规则取出来
//我们需要给producer配置ByteBufferSerializer,默认的kafkaTemplate是StringSerializer
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;

@Bean //纳入spring管理,不用每次去取值
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer.class);
    return props;
}
//我们需要给consumer配置ByteBufferSerializer,默认的kafkaTemplate是StringSerializer

@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;

@Value("${spring.kafka.consumer.group-id}")
private String consumerGroupId;

public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteBufferDeserializer.class);
    return props;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, ByteBuffer> eduTestKafkaFactory() {
    ConsumerFactory<String, ByteBuffer> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfigs());
    ConcurrentKafkaListenerContainerFactory<String, ByteBuffer> factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(consumerFactory);
    return factory;
}

//在KafkaListener的时候,配置containerFactory,就可以获取我们的consumerConfigs
@KafkaListener(topics = EDU_SUBMIT_TEST_TOPIC, containerFactory = "eduTestKafkaFactory")
通过我们自己的指定的serializer,我们可以把我们对象压缩为原来的十分之一,大大减小我们传给kafka的对象,当然我们当前只给指定类做了序列化,以后添加字段也不通用,现在可以采用Avro来(反)序列化对象。KafkaAvroDeserializer也是非常强大的。

问题

当线上考试的时候,每次提交就有可能导致成绩出不来,最开始初步猜想,数据丢失,可是数据为什么会丢失了?到底是kafka没有接受到数据,还是kafka接受到数据没有消费掉,我们采用的是kafkaListenner。还是有一部分提交可以成功的,开发和测试环境都是Ok的,只有生产有问题。

我们借助kafka的命令查看消费情况

kafka 查看所有的消费者组

bin/kafka-consumer-groups.sh --bootstrap-server ip:9092 --list
kafka 查看某个消费者组的消费数据情况
bin/kafka-consumer-groups.sh --bootstrap-server ip:9092 --group strive --describe
我们看到对同一个topic有一个消费组,但是这一个消费者有两个节点在执行,这个两个节点分别是生产和预生产,预生产是运营后台的,但是运营后台消费数据却没有写入到业务库里面,导致考试的成绩总是出不来。我们停掉预生产之后,考试就正常了,但是预生产还是需要启动的,于是我们根据环境来设置不同的groupId,并把对应的factory的autoStartUp设置为false

/**
 * 获取不同环境的消费者组Id
 * @return
 */
public String obtainConsumerGroupId() {
    String logEnv = environment.getProperty("log_env");
    String consumerGroupId = "";
    if (PRE_PROD.equals(logEnv))
        consumerGroupId = environment.getProperty("spring.application.name") + "-consumer-" + PRE_PROD;
    else
        consumerGroupId = environment.getProperty("spring.kafka.consumer.group-id");
    log.info("edu test kafka config curr consumer group id is [{}] .", consumerGroupId);
    return consumerGroupId;
}

/**
 * log_env=preprod  预生产 [预生产 不获取数据]
 * log_env=prod   生产
 * @return
 */
public boolean setAutoStartup() {
    String logEnv = environment.getProperty("log_env");
    log.info("edu test kafka config curr log_env [{}] .", logEnv);
    return !PRE_PROD.equals(logEnv);
}
 

思路:

削峰

优化内存和数据大小

持续观察

思考与讨论:

1.序列化和反序列化这块还没有完全理解,有机会可以再验证下。

2.在多内容传输情况下,可能会出现类似问题,可以从这方面排查。

3.去面试的时候面试我的小哥刚好也做过这块,有一些同感,不过他提出了一些不同的观点,都是设计服务器的,有以下几点:

1) 服务器请求数最高能接受多少

2) 服务器的最大带宽是多少

3) 压力高时,服务器哪块出现性能问题,是CPU、内存、还是IO,不同的指标可能说明不同的问题,比如是CPU,说明可能接口对应的程序处理逻辑复杂;如果是内存,可能是接口对应的方法内存管理不当,比如没有释放一些连接什么的;如果是IO,可能是接口对应的方法内有太多的IO操作,是否可能考虑更换为多路复用的NIO还是其它处理方式

我想了下他说的这几点也有些道理,后面再遇到类似问题也可以从这几方面入手,希望有机会能和他再请教和讨论下。

这篇关于曾经参与的一次并发优化回顾的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1055646

相关文章

MySQL索引的优化之LIKE模糊查询功能实现

《MySQL索引的优化之LIKE模糊查询功能实现》:本文主要介绍MySQL索引的优化之LIKE模糊查询功能实现,本文通过示例代码给大家介绍的非常详细,感兴趣的朋友一起看看吧... 目录一、前缀匹配优化二、后缀匹配优化三、中间匹配优化四、覆盖索引优化五、减少查询范围六、避免通配符开头七、使用外部搜索引擎八、分

Python通过模块化开发优化代码的技巧分享

《Python通过模块化开发优化代码的技巧分享》模块化开发就是把代码拆成一个个“零件”,该封装封装,该拆分拆分,下面小编就来和大家简单聊聊python如何用模块化开发进行代码优化吧... 目录什么是模块化开发如何拆分代码改进版:拆分成模块让模块更强大:使用 __init__.py你一定会遇到的问题模www.

Mybatis 传参与排序模糊查询功能实现

《Mybatis传参与排序模糊查询功能实现》:本文主要介绍Mybatis传参与排序模糊查询功能实现,本文通过实例代码给大家介绍的非常详细,感兴趣的朋友跟随小编一起看看吧... 目录一、#{ }和${ }传参的区别二、排序三、like查询四、数据库连接池五、mysql 开发企业规范一、#{ }和${ }传参的

SpringBoot首笔交易慢问题排查与优化方案

《SpringBoot首笔交易慢问题排查与优化方案》在我们的微服务项目中,遇到这样的问题:应用启动后,第一笔交易响应耗时高达4、5秒,而后续请求均能在毫秒级完成,这不仅触发监控告警,也极大影响了用户体... 目录问题背景排查步骤1. 日志分析2. 性能工具定位优化方案:提前预热各种资源1. Flowable

SpringBoot3实现Gzip压缩优化的技术指南

《SpringBoot3实现Gzip压缩优化的技术指南》随着Web应用的用户量和数据量增加,网络带宽和页面加载速度逐渐成为瓶颈,为了减少数据传输量,提高用户体验,我们可以使用Gzip压缩HTTP响应,... 目录1、简述2、配置2.1 添加依赖2.2 配置 Gzip 压缩3、服务端应用4、前端应用4.1 N

Spring Boot + MyBatis Plus 高效开发实战从入门到进阶优化(推荐)

《SpringBoot+MyBatisPlus高效开发实战从入门到进阶优化(推荐)》本文将详细介绍SpringBoot+MyBatisPlus的完整开发流程,并深入剖析分页查询、批量操作、动... 目录Spring Boot + MyBATis Plus 高效开发实战:从入门到进阶优化1. MyBatis

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.

MyBatis 动态 SQL 优化之标签的实战与技巧(常见用法)

《MyBatis动态SQL优化之标签的实战与技巧(常见用法)》本文通过详细的示例和实际应用场景,介绍了如何有效利用这些标签来优化MyBatis配置,提升开发效率,确保SQL的高效执行和安全性,感... 目录动态SQL详解一、动态SQL的核心概念1.1 什么是动态SQL?1.2 动态SQL的优点1.3 动态S

Python如何使用__slots__实现节省内存和性能优化

《Python如何使用__slots__实现节省内存和性能优化》你有想过,一个小小的__slots__能让你的Python类内存消耗直接减半吗,没错,今天咱们要聊的就是这个让人眼前一亮的技巧,感兴趣的... 目录背景:内存吃得满满的类__slots__:你的内存管理小助手举个大概的例子:看看效果如何?1.

一文详解SpringBoot响应压缩功能的配置与优化

《一文详解SpringBoot响应压缩功能的配置与优化》SpringBoot的响应压缩功能基于智能协商机制,需同时满足很多条件,本文主要为大家详细介绍了SpringBoot响应压缩功能的配置与优化,需... 目录一、核心工作机制1.1 自动协商触发条件1.2 压缩处理流程二、配置方案详解2.1 基础YAML