【Redis项目实战】使用Springcloud整合Redis分布式锁+RabbitMQ技术实现高并发预约管理处理系统

简介: 【Redis项目实战】使用Springcloud整合Redis分布式锁+RabbitMQ技术实现高并发预约管理处理系统

开发目的:


开发一个高并发预约管理处理系统,其中用户可以预约倾听者。由于并发预约可能导致冲突和混乱,需要实现分布式锁来确保同一时间段只有一个用户可以进行预约。为了实现这一目的,使用Redis作为分布式锁的存储介质,并设计一组表来存储倾听者的预约情况。


功能介绍:


  1. 高并发预约管理:系统能够处理大量用户同时预约倾听者的情况,通过使用分布式锁来保证同一时间段只有一个用户可以进行预约,防止冲突和混乱。


  1. 分布式锁实现:系统使用Redis作为分布式锁的存储介质,通过设置键值对来实现分布式锁。具体地,使用一组表来存储倾听者的预约情况,表名由倾听者的ID和日期组成。每个表使用Redis的哈希表结构,其中键表示时间段,值表示该时间段是否已被预约(真或假)。通过对这些表的操作,系统实现了分布式锁的效果。


  1. 预约冲突检测:在用户进行预约时,系统会检查对应倾听者的预约情况表,判断该时间段是否已被其他用户预约。如果时间段已被预约,则系统会阻止当前用户的预约请求,以避免冲突。


  1. 数据持久化:用户的预约信息会被保存到数据库中,以便后续查询和处理。同时,通过RabbitMQ等消息队列技术,系统可以将预约信息发送到其他模块进行处理。


技术实现步骤:


  • 系统中已经集成了Spring Cloud、Redis和RabbitMQ相关依赖。
  • 创建Redis分布式锁的工具类:
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.core.script.ScriptExecutor;
import org.springframework.data.redis.serializer.RedisSerializer;
 
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
 
public class RedisDistributedLock {
 
    private RedisTemplate<String, String> redisTemplate;
    private ScriptExecutor<String> scriptExecutor;
    private RedisSerializer<String> redisSerializer;
 
    public RedisDistributedLock(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.scriptExecutor = this.redisTemplate.getScriptExecutor();
        this.redisSerializer = this.redisTemplate.getStringSerializer();
    }
 
    /**
     * 尝试获取分布式锁
     *
     * @param lockKey     锁的key
     * @param requestId   锁的唯一标识
     * @param expireTime  锁的过期时间(单位:秒)
     * @return 是否成功获取锁
     */
    public boolean tryLock(String lockKey, String requestId, long expireTime) {
        RedisScript<Long> script = new DefaultRedisScript<>(
                "if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then " +
                        "return redis.call('expire', KEYS[1], ARGV[2]) " +
                        "else " +
                        "return 0 " +
                        "end",
                Long.class
        );
 
        Long result = scriptExecutor.execute(script, Collections.singletonList(lockKey), requestId, expireTime);
        return result != null && result == 1;
    }
 
    /**
     * 释放分布式锁
     *
     * @param lockKey   锁的key
     * @param requestId 锁的唯一标识
     */
    public void releaseLock(String lockKey, String requestId) {
        RedisScript<Long> script = new DefaultRedisScript<>(
                "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                        "return redis.call('del', KEYS[1]) " +
                        "else " +
                        "return 0 " +
                        "end",
                Long.class
        );
 
        scriptExecutor.execute(script, Collections.singletonList(lockKey), requestId);
    }
 
    /**
     * 生成唯一的锁标识
     *
     * @return 锁的唯一标识
     */
    public String generateRequestId() {
        return UUID.randomUUID().toString();
    }
}
  • 在预约服务中使用分布式锁来保证同一时间段只有一个用户可以进行预约:
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
 
@Service
public class ReservationService {
 
    private RedisDistributedLock distributedLock;
    private ReservationRepository reservationRepository;
    private RabbitTemplate rabbitTemplate;
 
    @Autowired
    public ReservationService(RedisDistributedLock distributedLock, ReservationRepository reservationRepository, RabbitTemplate rabbitTemplate) {
        this.distributedLock = distributedLock;
        this.reservationRepository = reservationRepository;
        this.rabbitTemplate = rabbitTemplate;
    }
 
    @Transactional
    public void makeReservation(String listenerId, String userId, String timeSlot) {
        String lockKey = listenerId + "-" + timeSlot;
        String requestId = distributedLock.generateRequestId();
 
        try {
            // 尝试获取分布式锁,设置锁的过期时间为一定的秒数(例如10秒)
            if (distributedLock.tryLock(lockKey, requestId, 10)) {
                // 获取到锁,可以进行预约操作
 
                // 检查时间段是否已经被预约
                if (!isTimeSlotBooked(listenerId, timeSlot)) {
                    // 保存预约信息到数据库
                    Reservation reservation = new Reservation(listenerId, userId, timeSlot);
                    reservationRepository.save(reservation);
 
                    // 发送预约消息到RabbitMQ
                    rabbitTemplate.convertAndSend("reservation-exchange", "reservation-queue", reservation);
                } else {
                    // 时间段已经被预约,抛出异常或返回错误信息
                    throw new RuntimeException("The time slot is already booked.");
                }
            } else {
                // 未获取到锁,抛出异常或返回错误信息
                throw new RuntimeException("Failed to acquire lock for the time slot.");
            }
        } finally {
            // 释放分布式锁
            distributedLock.releaseLock(lockKey, requestId);
        }
    }
 
    private boolean isTimeSlotBooked(String listenerId, String timeSlot) {
        // 查询数据库或Redis,判断时间段是否已经被预约
        // 返回true表示已被预约,返回false表示未被预约
    }
}
  • 在配置类中配置RedisTemplate和RabbitTemplate
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
@Configuration
public class AppConfig {
 
    private RedisConnectionFactory redisConnectionFactory;
    private ConnectionFactory rabbitConnectionFactory;
 
    @Autowired
    public AppConfig(RedisConnectionFactory redisConnectionFactory, ConnectionFactory rabbitConnectionFactory) {
        this.redisConnectionFactory = redisConnectionFactory;
        this.rabbitConnectionFactory = rabbitConnectionFactory;
    }
 
    @Bean
    public RedisTemplate<String, String> redisTemplate() {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new StringRedisSerializer());
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
 
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
        rabbitTemplate.setConnectionFactory(rabbitConnectionFactory);
        return rabbitTemplate;
    }
 
    @Bean
    public RedisDistributedLock distributedLock(RedisTemplate<String, String> redisTemplate) {
        return new RedisDistributedLock(redisTemplate);
    }
}

RabbitMQ是一种消息队列系统,可以用于异步处理预约消息,提高系统的可伸缩性和稳定性。以下是实现RabbitMQ部分的步骤:


  • 创建预约消息的实体类:
public class ReservationMessage {
    private String listenerId;
    private String userId;
    private String timeSlot;
 
    // 构造方法、Getter和Setter方法省略
}
  • 创建预约消息的消费者:
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
public class ReservationMessageConsumer {
 
    @RabbitListener(queues = "reservation-queue")
    public void processReservationMessage(ReservationMessage reservationMessage) {
        // 处理预约消息,例如发送通知给倾听者或用户
        System.out.println("Received reservation message: " + reservationMessage.toString());
    }
}
  • 在配置类中配置RabbitMQ相关的队列和交换机:
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitMQConfig {
 
    @Bean
    public Queue reservationQueue() {
        return new Queue("reservation-queue");
    }
 
    @Bean
    public DirectExchange reservationExchange() {
        return new DirectExchange("reservation-exchange");
    }
 
    @Bean
    public Binding reservationBinding(Queue reservationQueue, DirectExchange reservationExchange) {
        return BindingBuilder.bind(reservationQueue).to(reservationExchange).with("reservation-queue");
    }
}
  • 在预约服务中发送预约消息到RabbitMQ:
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
@Service
public class ReservationService {
 
    private RabbitTemplate rabbitTemplate;
 
    @Autowired
    public ReservationService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
 
    public void makeReservation(String listenerId, String userId, String timeSlot) {
        // 创建预约消息
        ReservationMessage reservationMessage = new ReservationMessage(listenerId, userId, timeSlot);
 
        // 发送预约消息到RabbitMQ
        rabbitTemplate.convertAndSend("reservation-exchange", "reservation-queue", reservationMessage);
    }
}


为了实现倾听者预约状态的管理,可以在Redis中建立一个临时状态表,与之前的分布式锁表类似。以下是该表的设计和状态介绍:


  1. 表名设计:使用倾听者的ID和日期作为表名,一次性生成七天的表。例如,表名可以是形如"listener_id_20240201"的格式。
  2. 状态类型:
  • 等待(Waiting):表示用户的预约请求已提交,但倾听者尚未对其进行处理。
  • 接受(Accepted):表示倾听者已接受用户的预约请求。
  • 拒绝(Rejected):表示倾听者已拒绝用户的预约请求。
  • 已处理(Processed):表示该预约请求已被处理,并且不再需要保留在状态表中。
  1. Redis表结构:使用Redis的哈希表结构来存储倾听者的预约状态。表中的键表示时间段,值表示对应时间段的预约状态。


以下是使用RedisTemplate实现倾听者预约状态管理的代码:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
import java.util.Map;
 
@Component
public class ReservationStatusManager {
 
    private static final String TABLE_PREFIX = "listener_";
 
    private RedisTemplate<String, Object> redisTemplate;
    private HashOperations<String, String, String> hashOperations;
 
    @Autowired
    public ReservationStatusManager(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
 
    @PostConstruct
    private void init() {
        hashOperations = redisTemplate.opsForHash();
    }
 
    public void setReservationStatus(String listenerId, String date, String timeSlot, String status) {
        String tableName = TABLE_PREFIX + listenerId + "_" + date;
        hashOperations.put(tableName, timeSlot, status);
    }
 
    public String getReservationStatus(String listenerId, String date, String timeSlot) {
        String tableName = TABLE_PREFIX + listenerId + "_" + date;
        return hashOperations.get(tableName, timeSlot);
    }
 
    public void removeProcessedReservations(String listenerId, String date) {
        String tableName = TABLE_PREFIX + listenerId + "_" + date;
        redisTemplate.delete(tableName);
    }
 
    public Map<String, String> getAllReservationStatus(String listenerId, String date) {
        String tableName = TABLE_PREFIX + listenerId + "_" + date;
        return hashOperations.entries(tableName);
    }
}


我们通过注入RedisTemplate来操作Redis。在ReservationStatusManager类中,我们使用HashOperations来进行哈希表的操作。init()方法使用@PostConstruct注解进行初始化,确保RedisTemplate和HashOperations在对象创建后进行实例化。


setReservationStatus()方法用于设置预约状态,getReservationStatus()方法用于获取预约状态,removeProcessedReservations()方法用于删除已处理的预约状态。另外,getAllReservationStatus()方法可以用于获取某个倾听者在特定日期的所有预约状态。


那么这样就实现了  在多个用户想要同时预约同一个倾听者的同一时间时 所避免的数据混乱


我们这样做保证了数据的一致性 又使用了 以速度性能著称的Redis 和异步处理的RabbitMQ这样就能使得 在预约成功的一瞬间 完成通知倾听者的效果 达到了预期的业务效果


相关文章
|
7月前
|
负载均衡 测试技术 调度
大模型分布式推理:张量并行与流水线并行技术
本文深入探讨大语言模型分布式推理的核心技术——张量并行与流水线并行。通过分析单GPU内存限制下的模型部署挑战,详细解析张量并行的矩阵分片策略、流水线并行的阶段划分机制,以及二者的混合并行架构。文章包含完整的分布式推理框架实现、通信优化策略和性能调优指南,为千亿参数大模型的分布式部署提供全面解决方案。
2240 4
|
Kubernetes 大数据 调度
Airflow vs Argo Workflows:分布式任务调度系统的“华山论剑”
本文对比了Apache Airflow与Argo Workflows两大分布式任务调度系统。两者均支持复杂的DAG任务编排、社区支持及任务调度功能,且具备优秀的用户界面。Airflow以Python为核心语言,适合数据科学家使用,拥有丰富的Operator库和云服务集成能力;而Argo Workflows基于Kubernetes设计,支持YAML和Python双语定义工作流,具备轻量化、高性能并发调度的优势,并通过Kubernetes的RBAC机制实现多用户隔离。在大数据和AI场景中,Airflow擅长结合云厂商服务,Argo则更适配Kubernetes生态下的深度集成。
1292 34
|
8月前
|
存储 算法 安全
“卧槽,系统又崩了!”——别慌,这也许是你看过最通俗易懂的分布式入门
本文深入解析分布式系统核心机制:数据分片与冗余副本实现扩展与高可用,租约、多数派及Gossip协议保障一致性与容错。探讨节点故障、网络延迟等挑战,揭示CFT/BFT容错原理,剖析规模与性能关系,为构建可靠分布式系统提供理论支撑。
379 2
|
8月前
|
消息中间件 监控 Java
Apache Kafka 分布式流处理平台技术详解与实践指南
本文档全面介绍 Apache Kafka 分布式流处理平台的核心概念、架构设计和实践应用。作为高吞吐量、低延迟的分布式消息系统,Kafka 已成为现代数据管道和流处理应用的事实标准。本文将深入探讨其生产者-消费者模型、主题分区机制、副本复制、流处理API等核心机制,帮助开发者构建可靠、可扩展的实时数据流处理系统。
780 4
|
8月前
|
监控 安全 Java
Spring Cloud 微服务治理技术详解与实践指南
本文档全面介绍 Spring Cloud 微服务治理框架的核心组件、架构设计和实践应用。作为 Spring 生态系统中构建分布式系统的标准工具箱,Spring Cloud 提供了一套完整的微服务解决方案,涵盖服务发现、配置管理、负载均衡、熔断器等关键功能。本文将深入探讨其核心组件的工作原理、集成方式以及在实际项目中的最佳实践,帮助开发者构建高可用、可扩展的分布式系统。
511 1
|
8月前
|
机器学习/深度学习 算法 安全
新型电力系统下多分布式电源接入配电网承载力评估方法研究(Matlab代码实现)
新型电力系统下多分布式电源接入配电网承载力评估方法研究(Matlab代码实现)
267 3
|
7月前
|
机器学习/深度学习 监控 PyTorch
68_分布式训练技术:DDP与Horovod
随着大型语言模型(LLM)规模的不断扩大,从早期的BERT(数亿参数)到如今的GPT-4(万亿级参数),单卡训练已经成为不可能完成的任务。分布式训练技术应运而生,成为大模型开发的核心基础设施。2025年,分布式训练技术已经发展到相当成熟的阶段,各种优化策略和框架不断涌现,为大模型训练提供了强大的支持。
976 0
|
10月前
|
数据采集 缓存 NoSQL
分布式新闻数据采集系统的同步效率优化实战
本文介绍了一个针对高频新闻站点的分布式爬虫系统优化方案。通过引入异步任务机制、本地缓存池、Redis pipeline 批量写入及身份池策略,系统采集效率提升近两倍,数据同步延迟显著降低,实现了分钟级热点追踪能力,为实时舆情监控与分析提供了高效、稳定的数据支持。
458 1
分布式新闻数据采集系统的同步效率优化实战
|
8月前
|
JSON 监控 Java
Elasticsearch 分布式搜索与分析引擎技术详解与实践指南
本文档全面介绍 Elasticsearch 分布式搜索与分析引擎的核心概念、架构设计和实践应用。作为基于 Lucene 的分布式搜索引擎,Elasticsearch 提供了近实时的搜索能力、强大的数据分析功能和可扩展的分布式架构。本文将深入探讨其索引机制、查询 DSL、集群管理、性能优化以及与各种应用场景的集成,帮助开发者构建高性能的搜索和分析系统。
556 0
|
8月前
|
Kubernetes Java 微服务
Spring Cloud 微服务架构技术解析与实践指南
本文档全面介绍 Spring Cloud 微服务架构的核心组件、设计理念和实现方案。作为构建分布式系统的综合工具箱,Spring Cloud 为微服务架构提供了服务发现、配置管理、负载均衡、熔断器等关键功能的标准化实现。本文将深入探讨其核心组件的工作原理、集成方式以及在实际项目中的最佳实践,帮助开发者构建高可用、可扩展的分布式系统。
705 0

热门文章

最新文章