使用分布式锁解决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。

相关文章

大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
关于数据埋点，你需要了解这些基本知识

关于数据埋点，你需要了解这些基本知识

产品汪每天都在和数据打交道，你知道数据来自哪里吗？ 移动app端内的用户行为数据大多来自埋点，了解一些埋点知识，能和数据分析师、技术侃大山，参与到前期的数据采集，更重要是让最终的埋点数据能为我所用，否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式，可以区分为： 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理：自动程度高的，能解决通用统计，便于统一化管理，但个性化定
阅读更多...
中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
好题——hdu2522（小数问题：求1/n的第一个循环节）

好题——hdu2522（小数问题：求1/n的第一个循环节）

好喜欢这题，第一次做小数问题，一开始真心没思路，然后参考了网上的一些资料。 知识点***********************************无限不循环小数即无理数，不能写作两整数之比*****************************（一开始没想到，小学没学好） 此题1/n肯定是一个有限循环小数，了解这些后就能做此题了。 按照除法的机制，用一个函数表示出来就可以了，代码如下
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2