使用分布式锁解决IM聊天数据重复插入的问题

2024-08-22 08:44
文章标签 数据 问题 使用 解决 重复 分布式 插入 聊天 im

本文主要是介绍使用分布式锁解决IM聊天数据重复插入的问题，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

导航

  • 业务背景
  • 问题分析与定位
  • 探索可行的解决方案
    • 数据库层面处理——唯一索引
    • 应用程序层面处理——分布式锁
  • 分布式锁概述
    • 分布式锁需要具备哪些特性？
    • 分布式锁有哪些实现方式？
      • 基于数据库的实现方式
      • 基于Redisson实现方式
  • Redission介绍
    • 概述
    • 可重入锁
  • 基于Redisson解决方案
    • 方案梳理
    • Springboot集成Redisson
  • 结语
  • 参考

本文首发《使用分布式锁解决IM聊天数据重复插入的问题》

业务背景和问题

在IM聊天业务中，除了自建聊天服务器，构架闭环的咨询聊天，往往还需要接入三方的平台的IM流量。

这个就不得不去适配各种平台的推流方式。

在我们自建的IM聊天服务解决方案中，IM会话创建和消息的接收是两个独立模块（接口）。
这种设计方式从客户端层面就将两个流程分开且保证了顺序性，有效避免了一些不可预知的问题。

但是，三方流量平台的是通过消息推流的方式将流量投递给我们，我们必须在接收流量的过程中完成客户、会话、消息的创建。



如果所有消息是排队，一个一个执行，那么这个流程是没有问题的。

但是，我们发现三方推送消息的时候偶尔会发生推送同一客户的多条消息的情况，这种并发写入，导致数据重复写入。

这种情况下，就可能会导致新客户创建多次，对应的会话也会创建多个。



而且还会带来数据查询中偶尔出现selectOne的异常。

desc":"org.mybatis.spring.MyBatisSystemException: nested exception is org.apache.ibatis.exceptions.TooManyResultsException: Expected one result (or null) to be returned by selectOne(), but found: 2

在没有查明具体问题之前，我们在特定查询的时候增加了limit 1限制，原则上取最新的那一条。

问题分析与定位

对于聊天场景来说，这种脏数据的产生是不能容忍的。

为了找到问题的根本解决办法，我们开始专项排查。

我把代码走读了一遍，发现代码层面没有明显的bug。但是，从数据上来看大概率是消息并发投递导致。

为了证明这种猜想，我编写了一个测试用例来验证。

具体做法，就是写一个python脚本程序，模拟10个线程，每个线程都会调用消息接收的业务接口，并且每个消息的fromUser和toUser都是一样的。

核心思想就是同时推送给一个人多条消息。

经过验证，数据重复写入的问题复现了。并发请求原因已经实锤。

这里给个简单示意图，解释一下并发请求的流程。



可行性方案探索

我们自己也思考了一下，大致的解决方案有两种：

  • 数据层面解决
  • 应用程序层面解决
数据层面解决

这个很好理解，利用Mysql字段唯一索引阻止重复插入，这是数据库自己的机制。
但是，因为user表中tenantUserId字段最初就为设计唯一索引。

ALTER TABLE user ADD UNIQUE uk_tenant_user_id( tenantUserId );

一旦为tenantUserId列加上唯一索引后，当上述并发情况发生时，请求1和请求2中必然有一者会优先完成数据的插入操作，而另一者则会得到类似错误。因此，最终保证user表中只有一条tenantUserId=xxx的记录存在。

 Cause: java.sql.SQLIntegrityConstraintViolationException: Duplicate entry 'xxx' for key 'tenantUserId'\n##

经过评估，目前单表已经仅仅2000w数据。短时间内升级不太现实。
而且历史数据的修复也不是一个小工程。

应用程序层面解决

另一种解决的思路是我们不依赖底层的数据库来为我们提供唯一性的保障，而是靠应用程序自身的代码逻辑来避免并发冲突。

之所以我们会遇到重复插入数据的问题，是因为“检测数据是否已经存在”和“插入数据”两个动作被分割开来。由于这两个步骤不具备原子性，才导致两个不同的请求可以同时通过第一步的检测。如果我们能够把这两个动作合并为一个原子操作，就可以避免数据冲突了。这时候我们就需要通过加锁，来实现这个代码块的原子性。


考虑到我们的应用程序API是多机部署的，我们决定采用业界比较成熟的分布式锁方案。

分布式锁概述

分布式锁需要具备哪些特性？
  • 在分布式系统环境下，同一时间只有一台机器的一个线程可以获取到锁
  • 高可用的获取锁与释放锁
  • 高性能的获取锁与释放锁
  • 具备可重入特性
  • 具备锁失效机制，防止死锁
  • 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败
分布式锁实现主要有如下三种：
  • 基于数据库实现分布式锁
  • 基于Zookeeper实现分布式锁
  • 基于Redis实现分布式锁

每种的具体实现可以参考《什么是分布式锁？实现分布式锁的三种方式》。

除了以上三种分布式锁实现以外，还有一种是基于Redission实现方式。
因为我们业务接口是基于Springboot框架，所以查阅了相关资料我们选择一种Redission实现。

Redission介绍

概述

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

以下是Redisson的结构：

Redisson作为独立节点 可以用于独立执行其他节点发布到分布式执行服务 和 分布式调度任务服务 里的远程任务。



可重入锁（Reentrant Lock）

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();

大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {try {...} finally {lock.unlock();}
}

Redisson同时还为分布式锁提供了异步执行的相关方法：

RLock lock = redisson.getLock("anyLock");
lock.lockAsync();
lock.lockAsync(10, TimeUnit.SECONDS);
Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);

RLock对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁，其他进程解锁则会抛出IllegalMonitorStateException错误。但是如果遇到需要其他进程也能解锁的情况，请使用分布式信号量Semaphore 对象.

关于Redisson的更多介绍请移步Redisson 中文文档

基于Redisson解决方案

在本案例中，我们采用了基于Redisson实现分布式锁的方式。

方案梳理

技术方案确定了，但是还是需要结合实际场景合理应用。
那么，我们在哪些环节加锁呢？



我们再次对消息接收处理流程进行梳理，在原来的基础上增加了分布式锁。

Springboot集成Redisson

pom.xml中引入redisson

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.34.1</version>
</dependency>

yml文件中redis配置

  redis:enabled: truehost: xxxxport: 6371password: xxxdatabase: 2timeout: 10000connectionPoolSize: 15connectionMinimumIdleSize: 5

redissonConfig.java

@Configuration
@ConditionalOnExpression("${spring.redis.enabled}")
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private String port;@Value("${spring.redis.timeout}")private String timeout;@Value("${spring.redis.password}")private String password;@Value("${spring.redis.database}")private int database;@Value("${spring.redis.connectionPoolSize}")private int connectionPoolSize;@Value("${spring.redis.connectionMinimumIdleSize}")private int connectionMinimumIdleSize;@Bean(name = "redissonClient")public RedissonClient redissonClient() {Config config = new Config();config.setCodec(new StringCodec());SingleServerConfig singleServerConfig =config.useSingleServer().setAddress("redis://" + host + ":" + port).setDatabase(database).setConnectionPoolSize(connectionPoolSize).setConnectionMinimumIdleSize(connectionMinimumIdleSize).setTimeout(Integer.parseInt(timeout));if (StringUtils.isNotBlank(password)) {singleServerConfig.setPassword(password);}return Redisson.create(config);}
}

上面准备好之后，就可以在使用了。

核心代码实现
        //新创建增加分布式锁String mutex = StrUtil.format("im:lock:user:{}", createUserDto.getTenantUserId());RLock lock = redissonClient.getLock(mutex);boolean successLock = lock.tryLock();if (!successLock) {// 获取分布式锁失败log.info(String.format("{\"Method\":\"%s\",\"content\":\"%s\"}", "【getOrCreateUser】", JsonUtil.toJson(createUserDto)));throw new BizException("该顾客已经在创建中了", ResponseCodeEnum.GET_R_LOCK_FAIL.getCode());}//创建用户User visitor = new User();visitor.setUserName(createUserDto.getTenantUserId());//...

        //消息创建过程中，首次创建顾客、会话，//在获取锁失败的情况下，增加重试机制try {receiveMessage(inputDto);}catch (BizException ex){log.error(String.format("{\"Method\":\"%s\",\"content\":\"%s\"}", "【receiveMessage】", ex));if(ex.getCode().equals(ResponseCodeEnum.GET_R_LOCK_FAIL.getCode())) {//重试一次Thread.sleep(1000);log.info(String.format("{\"Method\":\"%s\",\"content\":\"%s\"}", "【receiveMessage.retry】", JsonUtil.toJson(inputDto)));receiveMessage(inputDto);}}

Notes: 关于Springboot中如何使用Redisson,更加具体实现代码请移步《Spring Boot 实战纪实》，项目源码中可以查阅。

测试用例

确保写的代码是可调式的。-《对几次通宵加班发版的复盘和思考》

在这么多年的职业生涯中,我逐渐摸索出一个确保代码质量的笨方法——单步调试。

这里我们也写一个测试用例。具体是思路前面也提过，这里不再赘述。

import json
import requests
import time
import uuid
import threadingdef receive_xhs_msg():try:#请求urlurl = """http://localhost:7071/api/message/receive"""# 增加请求头headers = {"Content-Type": "application/json; charset=UTF-8"}message_id=str(uuid.uuid4())print('message_id:'+message_id)userInfo={"header_image":"xxx.jpg","nickname":"- ","user_id":"63038d28000000001200d311"}payload={"content":"6ED5KduMqTDJZ1ztw+ZPgw==~split~OMo7DD2gqsJqBafx9WKsZlnNNkcEYD4hLLPczczIFmr+YMtTB9Wz4ZI0MYCM4cF28kG7rfqnXdR9cRmamEJzHmKLfTmVxv5jzGUFVQOU00iimtunMAEJ4x76oJDrdAVUc4bJfV5zFLotz/Bm0WM9TADvD2cLhpHsVmaZRXaiJ96wMQgqx+K727l5S15jmMa5PiLqZqBO2q/G+WEkJSbfLQ==","from_user_id":"63038d28000000001200d311","intentCommentId":"","message_id":message_id,"message_source":2,"message_type":"HINT","timestamp":"1723268668573","to_user_id":"575d2c135e87e733f0162b88","user_info":[userInfo]}#转换成jsongetJson=json.dumps(payload)#构造发送请求response=requests.post(url=url,data=getJson,headers=headers)#打印响应数据print(response.text)time.sleep(1)except Exception as e:print('Error:',e)finally:print('执行完成')if __name__ == '__main__':threads = []for _ in range(10):  # 循环创建10个线程t = threading.Thread(target=receive_xhs_msg)threads.append(t)for t in threads:  # 循环启动10个线程t.start()t.join()

结语

分布式锁在日常工作中应用广泛，比如接口防抖(防重复提交)，并发处理等。

在近期的IM消息处理中，正好有了一次生动的实践。

一点浅浅的经验，分享给大家，希望能起到抛砖引玉的作用。

参考

  • Redisson 中文文档
  • 《什么是分布式锁？实现分布式锁的三种方式》
  • 《灵活运用分布式锁解决数据重复插入问题》

这篇关于使用分布式锁解决IM聊天数据重复插入的问题的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1095791。

相关文章

Java中流式并行操作parallelStream的原理和使用方法

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流（parallelStream）的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS
阅读更多...
Linux join命令的使用及说明

Linux join命令的使用及说明

《Linuxjoin命令的使用及说明》`join`命令用于在Linux中按字段将两个文件进行连接,类似于SQL的JOIN,它需要两个文件按用于匹配的字段排序,并且第一个文件的换行符必须是LF,`jo... 目录一. 基本语法二. 数据准备三. 指定文件的连接key四.-a输出指定文件的所有行五.-o指定输出
阅读更多...
Linux jq命令的使用解读

Linux jq命令的使用解读

《Linuxjq命令的使用解读》jq是一个强大的命令行工具,用于处理JSON数据,它可以用来查看、过滤、修改、格式化JSON数据,通过使用各种选项和过滤器,可以实现复杂的JSON处理任务... 目录一. 简介二. 选项2.1.2.2-c2.3-r2.4-R三. 字段提取3.1 普通字段3.2 数组字段四.
阅读更多...
Linux kill正在执行的后台任务 kill进程组使用详解

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本，并kill2.1 进程查看2.2 遇到的
阅读更多...
详解SpringBoot+Ehcache使用示例

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储：配置灵活性：编码示例引入依赖：配置ehcache.XML文件：配置
阅读更多...
Java 虚拟线程的创建与使用深度解析

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程？1.2 为什么需要虚拟线程？二、虚拟线程与平台线程对比代码对比示例：三
阅读更多...
Nginx分布式部署流程分析

Nginx分布式部署流程分析

《Nginx分布式部署流程分析》文章介绍Nginx在分布式部署中的反向代理和负载均衡作用,用于分发请求、减轻服务器压力及解决session共享问题,涵盖配置方法、策略及Java项目应用,并提及分布式事... 目录分布式部署NginxJava中的代理代理分为正向代理和反向代理正向代理反向代理Nginx应用场景
阅读更多...
k8s按需创建PV和使用PVC详解

k8s按需创建PV和使用PVC详解

《k8s按需创建PV和使用PVC详解》Kubernetes中,PV和PVC用于管理持久存储,StorageClass实现动态PV分配,PVC声明存储需求并绑定PV,通过kubectl验证状态,注意回收... 目录1.按需创建 PV（使用 StorageClass）创建 StorageClass2.创建 PV
阅读更多...
IDEA和GIT关于文件中LF和CRLF问题及解决

IDEA和GIT关于文件中LF和CRLF问题及解决

《IDEA和GIT关于文件中LF和CRLF问题及解决》文章总结：因IDEA默认使用CRLF换行符导致Shell脚本在Linux运行报错,需在编辑器和Git中统一为LF,通过调整Git的core.aut... 目录问题描述问题思考解决过程总结问题描述项目软件安装shell脚本上git仓库管理，但拉取后，上l
阅读更多...
Redis 基本数据类型和使用详解

Redis 基本数据类型和使用详解

《Redis基本数据类型和使用详解》String是Redis最基本的数据类型,一个键对应一个值,它的功能十分强大,可以存储字符串、整数、浮点数等多种数据格式,本文给大家介绍Redis基本数据类型和... 目录一、Redis 入门介绍二、Redis 的五大基本数据类型2.1 String 类型2.2 Hash
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2