使用分布式锁解决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: true host: xxxx port: 6371 password: xxx database: 2 timeout: 10000 connectionPoolSize: 15 connectionMinimumIdleSize: 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 jsonimport requestsimport timeimport uuidimport threadingdef receive_xhs_msg(): try: #请求url url ="""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] } #转换成json getJson=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消息处理中，正好有了一次生动的实践。

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

参考

From:https://www.cnblogs.com/lucky_hu/p/18372851
博客园
100+评论
captcha
    相关文章
    1. ·8月前
      Dapr v1.14 版本已发布
      Dapr是一套开源、可移植的事件驱动型运行时，允许开发人员轻松立足云端与边缘位置运行弹性、微服务、无状态以及有状态等应用程序类型。Dapr能够确保开发人员专注于编写业务逻辑，而不必分神于解决分布式系统难题，由此显著提高生产力并缩短开发时长。Dapr 是用于构建云原生应用程序的开发人员框架，可以更轻松
      编程知识
    2. ·8月前
      使用FModel提取黑神话悟空的资产
      目录前言设置效果展示闲聊可能遇到的问题没有相应的UE引擎版本选项 前言 黑神话悟空昨天上线了，解个包looklook。 本文内容比较简洁，仅介绍解包黑神话所需的专项配置，关于FModel的基础使用流程，请见《使用FModel提取UE4/5游戏资产》 本文仅演示steam平台下的解包过程 设置 在FM
      编程知识
    3. ·8月前
      Python被远程主机强制关闭后怎么自动重新运行进程
      要实现Python程序在被远程主机强制关闭后能够自动重新运行，我们可以采用几种方法，但最直接且常用的方法之一是结合操作系统级的工具或脚本。在Linux系统中，我们可以使用cron作业或者systemd服务来实现这一功能；在Windows系统中，可以使用任务计划程序。但在这里，为了提供一个跨平台的、更
      编程知识
    4. ·8月前
      CVSS（Common Vulnerability Scoring System）打分规则解读
      CVSS（Common Vulnerability Scoring System）提供了一种根据漏洞的主要特征进行打分，反映其严重性的方法。CVSS 已成为被广泛使用的标准。 下面是CVSS 3.1版本计算器的界面截图，本文对Base Score的打分标准做解读，并提供一些建议。同时会对每个维度选项
      编程知识
    5. ·8月前
      DDD建模后写代码的正确姿势（Java、dotnet双平台）
      本文书接上回《一种很变态但有效的DDD建模沟通方式》，关注公众号（老肖想当外语大佬）获取信息： 最新文章更新； DDD框架源码（.NET、Java双平台）； 加群畅聊，建模分析、技术交流； 视频和直播在B站。 终于到了写代码的环节 如果你已经阅读过本系列前面的所有文章，我相信你对需求分析和建模设计有
      编程知识
    6. ·8月前
      从源码分析 SpringBoot 的 LoggingSystem → 它是如何绑定日志组件的
      开心一刻 今天心情不好，想约哥们喝点 我：心情不好，给你女朋友说一声，来我家，过来喝点 哥们：行！我给她说一声 我：你想吃啥？我点外卖 哥们：你俩定吧，我已经让她过去了 我：？？？我踏马让你过来！和她说一声 哥们：哈哈哈，我踏马寻思让她过去呢 前情回顾 SpringBoot2.7 霸王硬上弓 Log
      编程知识
    7. ·8月前
      一个能够生成 Markdown 表格的 Bash 脚本
      哈喽大家好，我是咸鱼。 今天分享一个很实用的 bash 脚本，可以通过手动提供单元格内容和列数或者将带有分隔符的文件（如 CSV、TSV 文件）转换为 Markdown 表格。 源代码在文末哦！原文链接：https://josh.fail/2022/pure-bash-markdown-table-
      编程知识
    8. ·8月前
      聊聊 PHP 多进程模式下的孤儿进程和僵尸进程
      在 PHP 的编程实践中多进程通常都是在 cli 脚本的模式下使用，我依稀还记得在多年以前为了实现从数据库导出千万级别的数据，第一次在 PHP 脚本中采用了多进程编程。
      编程知识