SpringBoot+Redis+RabbitMQ完成增删改查

各部分分工职责

RabbitMQ负责添加、修改、删除的异步利用
Redis负责数据的缓存
RabbitMQ里面角色职责简单描述

RabbitMQ里面有几个角色要先分清以及他们的对应关系：
交换机、队列、路由键
交换机和队列是一对多
队列和路由键是多对多
然后就是消息的发送者（生产者）和消息的继承者（消耗者）
此案例中，添加修改删除要从生产者发送到消耗者，也就是说，消耗者才是具体干活的角色，消息生产者只需要把消息发送到对应的队列中，由交换机根据路由键发送到对应的队列中
Redis职责简单描述

Redis只需要把要看的数据以及新添加的数据，添加到缓冲中即可，如果缓冲中没有，就从数据库查，再添加到缓存中，所以此次数据类型用的Hash
pom.xml文件坐标引入

1.         <!-- redis工具 -->
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter-data-redis</artifactId>
5.         </dependency>
6.         <!-- JSON工具 https://mvnrepository.com/artifact/com.alibaba/fastjson -->
7.         <dependency>
8.             <groupId>com.alibaba</groupId>
9.             <artifactId>fastjson</artifactId>
10.             <version>1.2.47</version>
11.         </dependency>
12.         <!--spring2.X集成redis所需common-pool2-->
13.         <dependency>
14.             <groupId>org.apache.commons</groupId>
15.             <artifactId>commons-pool2</artifactId>
16.             <version>2.6.0</version>
17.         </dependency>
19.         <!-- Spring Boot Starter AMQP -->
20.         <dependency>
21.             <groupId>org.springframework.boot</groupId>
22.             <artifactId>spring-boot-starter-amqp</artifactId>
23.         </dependency>

复制代码

SpringBoot的设置文件

1. # RabbitMQ:配置,服务器地址,端口,用户名,密码
2. spring.rabbitmq.host=localhost
3. spring.rabbitmq.port=5672
4. spring.rabbitmq.username=guest
5. spring.rabbitmq.password=guest
8. # 使用 Redis 作为缓存存储,具体配置:服务器地址,端口,密码
9. spring.cache.type=redis
10. spring.redis.host=127.0.0.1
11. spring.redis.port=6379
12. #spring.redis.password=root
13. spring.redis.password=
14. # 连接工厂使用的数据库索引,redis默认有16个db,索引0-15
15. spring.redis.database=0
16. #spring.redis.timeout=0
17. # 连接池最大连接数（使用负值表示没有限制） 这个值决定了同时可以有多少个活动的连接
18. spring.redis.lettuce.pool.max-active=8
19. ## 连接池最大阻塞等待时间（-1表示没有限制） 当连接池中的所有连接都被占用时，新的请求会等待一段时间
20. spring.redis.lettuce.pool.max-wait=-1
21. ## 连接池中的最大空闲连接，连接池中最多可以有多少个空闲的连接
22. spring.redis.lettuce.pool.max-idle=8
23. ## 连接池中的最小空闲连接，连接池中至少要有多少个空闲的连接
24. spring.redis.lettuce.pool.min-idle=0

复制代码

Redis设置类

1. @EnableCaching      //开启缓存
2. @Configuration
3. public class RedisConfig {
5.     @Bean
6.     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
7.         RedisTemplate<String, Object> template = new RedisTemplate<>();
8.         template.setConnectionFactory(factory);
9.         // 使用 Jackson2JsonRedisSerializer 来序列化和反序列化 Redis 的值（Value）
10.         GenericJackson2JsonRedisSerializer jackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
11.         // 使用 StringRedisSerializer 来序列化和反序列化 Redis 的键（Key）
12.         StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
13.         // 设置键（key）的序列化器
14.         template.setKeySerializer(stringRedisSerializer);
15.         // 设置值（value）的序列化器
16.         template.setValueSerializer(jackson2JsonRedisSerializer);
17.         // 设置 Hash 键（key）的序列化器
18.         template.setHashKeySerializer(stringRedisSerializer);
19.         // 设置 Hash 值（value）的序列化器
20.         template.setHashValueSerializer(jackson2JsonRedisSerializer);
22.         template.afterPropertiesSet();
23.         return template;
24.     }
26.     @Bean
27.     public CacheManager cacheManager(RedisConnectionFactory factory) {
28.         RedisSerializer<String> redisSerializer = new StringRedisSerializer();
29.         Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
30.         //解决查询缓存转换异常的问题
31.         ObjectMapper om = new ObjectMapper();
32.         om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
33.         om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
34.         jackson2JsonRedisSerializer.setObjectMapper(om);
35.         // 配置序列化（解决乱码的问题）,过期时间600秒
36.         RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
37.                 .entryTtl(Duration.ofSeconds(600))  //设置数据过期时间600秒
38.                 .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
39.                 .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
40.                 .disableCachingNullValues();
41.         RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
42.                 .cacheDefaults(config)
43.                 .build();
44.         return cacheManager;
45.     }
46. }

复制代码

RabbitMQ设置类

如果没有设置logback，就把log.info()的代码全部删了，不影响运行，后面有的话也都删了

1. @Configuration
2. public class RabbitMQConfig {
4.     public static final String EXCHANGE_NAME = "bill.exchange"; // 交换机名称
5.     public static final String QUEUE_SAVE_UPDATE = "bill.saveupdate"; // 保存修改队列
6.     public static final String QUEUE_DELETE = "bill.delete"; // 删除队列
7.     public static final String ROUTING_SAVE_UPDATE_KEY = "bill.saveupdatekey"; // 保存修改路由键
8.     public static final String ROUTING_DELETE_KEY = "bill.deletekey"; // 删除路由键
10.     /**
11.      * 添加/修改定义队列
12.      * @return 队列对象
13.      */
14.     @Bean
15.     public Queue queueSaveUpdate() {
16.         log.info(QUEUE_SAVE_UPDATE + "：RabbitMQ队列初始化成功：" + LocalDateTime.now());
17.         return new Queue(QUEUE_SAVE_UPDATE, true); // durable: 是否持久化
18.     }
20.     /**
21.      * 删除定义队列
22.      * @return 队列对象
23.      */
24.     @Bean
25.     public Queue queueDelete() {
26.         log.info(QUEUE_DELETE + "：RabbitMQ队列初始化成功：" + LocalDateTime.now());
27.         return new Queue(QUEUE_DELETE, true); // durable: 是否持久化
28.     }
30.     /**
31.      * 定义交换机
32.      * @return 交换机对象
33.      */
34.     @Bean
35.     public TopicExchange exchange() {
36.         log.info(EXCHANGE_NAME + "：RabbitMQ交换机初始化成功：" + LocalDateTime.now());
37.         return new TopicExchange(EXCHANGE_NAME);
38.     }
40.     /**
41.      * 绑定队列和交换机
42.      * @return 绑定对象
43.      */
44.     @Bean
45.     public Binding bindingSaveUpdate() {
46.         log.info(ROUTING_SAVE_UPDATE_KEY + "：RabbitMQ绑定队列和交换机成功：" + LocalDateTime.now());
47.         return BindingBuilder.bind(queueSaveUpdate()).to(exchange()).with(ROUTING_SAVE_UPDATE_KEY);
48.     }
50.     /**
51.      * 绑定队列和交换机
52.      * @return 绑定对象
53.      */
54.     @Bean
55.     public Binding bindingDelete() {
56.         log.info(ROUTING_DELETE_KEY + "：RabbitMQ绑定队列和交换机成功：" + LocalDateTime.now());
57.         return BindingBuilder.bind(queueDelete()).to(exchange()).with(ROUTING_DELETE_KEY);
58.     }
59. }

复制代码

RabbitMQ设置消息发送者（生产者）

也就是说，在需要异步调用的地方，注入BillMessageSender，然后，调对应的方法就可以了

1. @Service
2. public class BillMessageSender {
4.     @Autowired
5.     private RabbitTemplate rabbitTemplate;
7.     /**
8.      * 发送用户添加修改消息
9.      * @param bill 参数对象
10.      */
11.     public void sendBillSaveUpdateMessage(Bill bill) {
12.         rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_SAVE_UPDATE_KEY, bill);
13.     }
15.     /**
16.      * 发送用户删除消息
17.      * @param ids 参数列表
18.      */
19.     public void sendBillDeleteMessage(List<Long> ids) {
20.         rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_DELETE_KEY, ids);
21.     }
22. }

复制代码

异步发送消息

这是在Service层，
所以把ApiResult()这个自定义返回类换成你们的就行了
ObjectMapper这个工具重要是用来处理JSON数据的，这里我用是由于为了方便实体类和Map之间相互转换的，BillMapper是我自己用到的，这个可以换成你们自己的，不影响，剩下的就基本上没啥了，有不懂的可以批评区问，看到会回复

1. @Slf4j
2. @Service
3. public class BillRedisService {
5.     @Resource
6.     private BillMapper billMapper;
8.     @Resource
9.     private ObjectMapper objectMapper;
11.     @Resource
12.     private BillMessageSender billMessageSender;
14.     @Resource
15.     private RedisTemplate<String, Object> redisTemplate;
17.     // redis 中 bill的键
18.     private static final String BILL_REDIS_KEY = "bill:info";
21.     /**
22.      * 保存或者修改信息
23.      * @param bill
24.      * @return
25.      */
26.     @Transactional
27.     public ApiResult saveUpdateByRedis(Bill bill){
28.         if(bill.getId() == null){
29.             billMessageSender.sendBillSaveUpdateMessage(bill);   //把要添加的信息放入消息队列
30.         }else {
31.             log.info("[ " + bill.getId() + " ] 修改缓存中的数据");
32.             String key = BILL_REDIS_KEY + bill.getId();          //找到要修改值对应的redis的key
33.             Map map = objectMapper.convertValue(bill, Map.class);//把对象转换成map
34.             redisTemplate.opsForHash().putAll(key, map);         //更新数据到缓存中
35.             billMessageSender.sendBillSaveUpdateMessage(bill);   //把要修改的信息放入消息队列
36.         }
37.         return ApiResult.success();
38.     }
40.     /**
41.      * 从缓存取数据
42.      * @param id
43.      * @return
44.      */
45.     @Transactional
46.     public Bill selectPrimaryKeyByRedis(Long id){
47.         Bill bill;
48.         String key = BILL_REDIS_KEY + id;
49.         //有这个键就取数据，不然就查数据库
50.         if (redisTemplate.hasKey(key)) {
51.             log.info("[ " + id + " ] 从缓存中取数据");
52.             Map<Object, Object> map = redisTemplate.opsForHash().entries(key);
53.             bill = objectMapper.convertValue(map, Bill.class);
54.         }else{
55.             log.info("[ " + id + " ] 缓存中没有，向数据库中查询数据");
56.             bill = billMapper.selectByPrimaryKey(id);
57.             String putKey = BILL_REDIS_KEY + bill.getId();          //找到要修改值对应的redis的key
58.             Map map = objectMapper.convertValue(bill, Map.class);   //把对象转换成map
59.             redisTemplate.opsForHash().putAll(putKey, map);         //更新数据到缓存中
60.         }
61.         return bill;
62.     }
64.     /**
65.      * 删除方法
66.      * @param ids
67.      * @return
68.      */
69.     @Transactional
70.     public ApiResult delByRedis(List<Long> ids) {
71.         log.info("[ " + Arrays.toString(ids.toArray()) + " ] 以上数据要被删除");
72.         for (Long id : ids) {
73.             String key = BILL_REDIS_KEY + id;
74.             redisTemplate.delete(key);  //删除缓存中的数据
75.         }
76.         //数据库信息交给消息队列删除
77.         billMessageSender.sendBillDeleteMessage(ids);
78.         return ApiResult.success();
79.     }
80. }

复制代码

RabbitMQ设置消息接收者（消耗者）

添加上这个注解@RabbitListener(queues = RabbitMQConfig.QUEUE_SAVE_UPDATE)，
并指明监听的队列queues = RabbitMQConfig.QUEUE_SAVE_UPDATE
就能获取到消息发送者发送过来的使命以及使命参数了，就可以在这里写处理逻辑了，如果没有设置logback，可以把@Slf4j，以及log.info()，这两个代码删除了

1. @Slf4j
2. @Component
3. public class BillMessageReceiver {
5.     @Autowired
6.     private BillMapper billMapper;
7.     @Autowired
8.     private ObjectMapper objectMapper;
9.     @Resource
10.     private RedisTemplate<String, Object> redisTemplate;
12.     // redis 中 bill的键
13.     private static final String BILL_REDIS_KEY = "bill:info";
15.     /**
16.      * 处理添加和修改操作
17.      * @param bill 参数对象
18.      */
19.     @Transactional
20.     @RabbitListener(queues = RabbitMQConfig.QUEUE_SAVE_UPDATE)
21.     public void receiveBillSaveUpdateMessage(Bill bill) {
22.         log.info(RabbitMQConfig.QUEUE_SAVE_UPDATE + " 队列获取到数据：" + bill.toString());
23.         if (bill == null) {
24.             return;
25.         }
26.         if (bill.getId() == null) {
27.             bill.setDeleted(0);
28.             bill.setCreateTime(new Date());
29.             billMapper.insertSelective(bill);
30.         } else {
31.             bill.setUpdateTime(new Date());
32.             billMapper.updateByPrimaryKeySelective(bill);
33.         }
34.         String key = BILL_REDIS_KEY + bill.getId();          //添加后就有主键了,拼接成redis的key
35.         Map map = objectMapper.convertValue(bill, Map.class);//把对象转换成map
36.         redisTemplate.opsForHash().putAll(key, map);         //把添加的数据放到缓存中
37.     }
39.     /**
40.      * 删除数据
41.      * @param ids 参数列表
42.      */
43.     @Transactional
44.     @RabbitListener(queues = RabbitMQConfig.QUEUE_DELETE)
45.     public void receiveBillDeleteMessage(List<Long> ids) {
46.         log.info(RabbitMQConfig.QUEUE_DELETE + " 队列获取到数据：" + Arrays.toString(ids.toArray()));
47.         Bill bill = new Bill();
48.         bill.setDeleted(1);
50.         Example example = new Example(Bill.class);
51.         Example.Criteria criteria = example.createCriteria();
52.         criteria.andIn("id",ids);
54.         billMapper.updateByExampleSelective(bill,example);
55.     }
57. }

复制代码

总结

再把逻辑捋一下
添加修改删除，这些利用同一发送给RabbitMQ，由RabbitMQ的消耗者处理后续利用
查看详情，添加和更新的数据，交给Redis缓存，缓存没有，就查数据库，然后再缓存到Redis中，就第一遍查数据库，后续走的都是缓存
以上代码实现的功能就是，
全部数据查询还是走的数据库（数据量不多），但是单个查询，查询详情，先查缓存，缓存没有再查数据库，然后再添加到缓存中，下次查询就不走数据库了
添加修改删除同一发送给RabbitMQ消息队列，由消息队列异步完成后续的使命，并更新大概删除对应的缓存
这比之前单独的对数据库利用，多了2层逻辑，RabbitMQ和缓存的处理，这个例子就是简单的使用RabbitMQ和Redis，算是个小入门，如果有其他好的建议，可以批评一下，非常感谢！


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。