springboot2.X 集成redis+消息发布订阅

需求场景：分布式项目中，每个子项目有各自的 user 数据库， 在综合管理系统中存放这所有用户信息， 为了保持综合管理系统用户的完整性，

　　　　子系统添加用户后将用户信息以json格式保存至redis，然后发布到消息到消息通道，综合管理系统监控到子系统发布的消息前往redis 

　　　　获取出用户信息保存到自己的数据库

1）redis配置

1 1 spring:
2 2   redis:
3 3     #数据库索引
4 4     database: 5
5 7     host: 127.0.0.1
6 8     port: 6379
7 9     password: 123456
810     jedis:
911       pool:
1012         #最大连接数
1113         max-active: 8
1214         #最大阻塞等待时间(负数表示没限制)
1315         #最大空闲
1416         max-idle: 8
1517         #最小空闲
1618         min-idle: 0

2）集成redis ， 初始化redis组件

1  1 package com.bigcustomer.configs;
2  2 
3  3 
4  4 import com.bigcustomer.utils.redisUtil.RedisService;
5  5 import com.fasterxml.jackson.annotation.JsonAutoDetect;
6  6 import com.fasterxml.jackson.annotation.PropertyAccessor;
7  7 import com.fasterxml.jackson.databind.ObjectMapper;
8  8 import org.slf4j.Logger;
9  9 import org.slf4j.LoggerFactory;
10 10 import org.springframework.beans.factory.annotation.Autowired;
11 11 import org.springframework.cache.annotation.CachingConfigurerSupport;
12 12 import org.springframework.cache.annotation.EnableCaching;
13 13 import org.springframework.context.annotation.Bean;
14 14 import org.springframework.context.annotation.Configuration;
15 15 import org.springframework.data.redis.connection.RedisConnectionFactory;
16 16 import org.springframework.data.redis.core.RedisTemplate;
17 17 import org.springframework.data.redis.core.StringRedisTemplate;
18 18 import org.springframework.data.redis.listener.PatternTopic;
19 19 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
20 20 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
21 21 import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
22 22 
23 23 
24 24 /**
25 25  * @author :CX
26 26  * @Date :Create in 2018/8/15 14:03
27 27  * @Effect :
28 28  */
29 29 
30 30 @Configuration
31 31 @EnableCaching//开启注解
32 32 public class RedisConfig extends CachingConfigurerSupport {
33 33 
34 34 
35 35     private static Logger logger = LoggerFactory.getLogger(RedisConfig.class);
36 36     // 自定义的配置类， 存放了通道地址
37 37     @Autowired
38 38     private BaseConfig baseConfig;
39 39 
40 40     /**
41 41      *@参数
42 42      *@返回值
43 43      *@创建人  cx
44 44      *@创建时间
45 45      *@描述   //初始化监听器
46 46      */
47 47     @Bean
48 48     RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
49 49                                             MessageListenerAdapter listenerAdapter) {
50 50 
51 51         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
52 52         container.setConnectionFactory(connectionFactory);
53 53         //配置监听通道
54 54         container.addMessageListener(listenerAdapter, new PatternTopic(baseConfig.getRedisAisle()));// 通道的名字
55 55         logger.info("初始化监听成功，监听通道：【"+baseConfig.getRedisAisle()+"】");
56 56         return container;
57 57     }
58 58 
59 59     /**
60 60      *@参数
61 61      *@返回值
62 62      *@创建人  cx
63 63      *@创建时间
64 64      *@描述 利用反射来创建监听到消息之后的执行方法
65 65      */
66 66     @Bean
67 67     MessageListenerAdapter listenerAdapter(RedisService receiver) {
68 68         return new MessageListenerAdapter(receiver, "receiveMessage");
69 69     }
70 70 
71 71 //    /**
72 72 //     *@参数
73 73 //     *@返回值
74 74 //     *@创建人  cx
75 75 //     *@创建时间
76 76 //     *@描述 控制线程用的
77 77 //     */
78 78 //    @Bean
79 79 //    Receiver receiver(CountDownLatch latch) {
80 80 //        return new Receiver(latch);
81 81 //    }
82 82 //
83 83 //    @Bean
84 84 //    CountDownLatch latch() {
85 85 //        return new CountDownLatch(1);
86 86 //    }
87 87 
88 88     /**
89 89      *@参数
90 90      *@返回值
91 91      *@创建人  cx
92 92      *@创建时间
93 93      *@描述   //使用默认的工厂初始化redis操作String模板
94 94      */
95 95     @Bean
96 96     StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
97 97         return new StringRedisTemplate(connectionFactory);
98 98     }
99 99     /**
100100      *@参数
101101      *@返回值
102102      *@创建人  cx
103103      *@创建时间
104104      *@描述   //使用默认的工厂初始化redis操作map模板
105105      */
106106     @Bean
107107     RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory) {
108108 
109109         Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
110110         ObjectMapper om = new ObjectMapper();
111111         om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
112112         om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
113113         jackson2JsonRedisSerializer.setObjectMapper(om);
114114         RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
115115         template.setConnectionFactory(connectionFactory);
116116         template.setKeySerializer(jackson2JsonRedisSerializer);
117117         template.setValueSerializer(jackson2JsonRedisSerializer);
118118         template.setHashKeySerializer(jackson2JsonRedisSerializer);
119119         template.setHashValueSerializer(jackson2JsonRedisSerializer);
120120         template.afterPropertiesSet();
121121         return template;
122122 
123123     }
124124 }

3 ） 操作string 和 map 的dao封装

1  1 package com.bigcustomer.utils.redisUtil;
2  2 
3  3 import com.alibaba.fastjson.JSON;
4  4 import com.bigcustomer.biguser.service.BigUserService;
5  5 import org.slf4j.Logger;
6  6 import org.slf4j.LoggerFactory;
7  7 import org.springframework.data.redis.core.RedisTemplate;
8  8 import org.springframework.data.redis.core.StringRedisTemplate;
9  9 import org.springframework.stereotype.Component;
10 10 
11 11 import javax.annotation.Resource;
12 12 import java.util.Map;
13 13 
14 14 /**
15 15  * @author :CX
16 16  * @Date :Create in 2018/8/15 14:19
17 17  * @Effect : redisDAO封裝
18 18  */
19 19 @Component
20 20 public class MyRedisDao {
21 21 
22 22     private static Logger logger = LoggerFactory.getLogger(BigUserService.class);
23 23     @Resource
24 24     private StringRedisTemplate template;
25 25 
26 26     @Resource
27 27     private RedisTemplate redisTemplate;
28 28 
29 29     //大客户信息同步到redis时保存的map的key
30 30     private final String BIG_USER_REDIS_KEY = "CM:CHANNELCUSTOMER";
31 31 
32 32     /**
33 33      * @参数
34 34      * @返回值
35 35      * @创建人 cx
36 36      * @创建时间
37 37      * @描述 大客户添加成功后存到redis
38 38      */
39 39     public boolean setMap(Map<String , Object> map) {
40 40 
41 41         try {
42 42             redisTemplate.opsForHash().putAll(BIG_USER_REDIS_KEY
43 43                     , map);
44 44             logger.info("同步大客户信息到redis 成功！userId【" + map.get("funiqueid")+ "】");
45 45             return true;
46 46         } catch (Exception e) {
47 47             e.printStackTrace();
48 48         }
49 49         logger.info("同步大客户信息到redis 失败！userId【" + map.get("funiqueid")+ "】");
50 50         return false;
51 51     }
52 52 
53 53 
54 54     public Object getMap(String key) {
55 55 
56 56         try {
57 57             Object o = redisTemplate.opsForHash().get(BIG_USER_REDIS_KEY, key);
58 58             if (null != o) {
59 59                 return o;
60 60             }
61 61         } catch (Exception e) {
62 62             e.printStackTrace();
63 63         }
64 64         logger.info("获取大客户信息到失败！");
65 65         return null;
66 66     }
67 67 
68 68 
69 69     /**
70 70      * @参数
71 71      * @返回值 存在 = true , 不纯在false
72 72      * @创建人 cx
73 73      * @创建时间
74 74      * @描述 判断是否存在 该key对应的值
75 75      */
76 76     public boolean isNull(String key) {
77 77         return template.hasKey(key);
78 78     }
79 79 
80 80     /**
81 81      * @参数
82 82      * @返回值
83 83      * @创建人 cx
84 84      * @创建时间
85 85      * @描述 设置值 和 过期时间 单位秒
86 86      */
87 87     public boolean setValue(String key, Object val, long expire) {
88 88         if (!this.isNull(key)) {
89 89             //不存在
90 90             String jsonString = JSON.toJSONString(val);
91 91             template.opsForValue().set(key, jsonString, expire);
92 92             logger.info("***************************成功在缓存中插入：" + key);
93 93             return true;
94 94         } else {
95 95             logger.info("***************************【" + key + "】已经存在缓存");
96 96             return false;
97 97         }
98 98     }
99 99 
100100 
101101     /**
102102      * @参数
103103      * @返回值
104104      * @创建人 cx
105105      * @创建时间
106106      * @描述 删除
107107      */
108108     public boolean del(String key) {
109109         return template.delete(key);
110110     }
111111 
112112     /**
113113      * @参数
114114      * @返回值
115115      * @创建人 cx
116116      * @创建时间
117117      * @描述 插入直接覆盖
118118      */
119119     public boolean setValue(String key, Object val) {
120120         //不存在
121121         String jsonString = JSON.toJSONString(val);
122122         // 去掉多余的 “
123123         String replace = jsonString.replace("\"", "");
124124         template.opsForValue().set(key, replace);
125125         logger.info("***************************成功在缓存中插入：" + key);
126126         return true;
127127     }
128128 
129129     /**
130130      * @参数
131131      * @返回值
132132      * @创建人 cx
133133      * @创建时间
134134      * @描述 获取对应key 的值
135135      */
136136     public String getValue(String key) {
137137         if (!this.isNull(key)) {
138138 
139139             //不存在
140140             logger.info("***************************【" + key + "】不存在缓存");
141141             return null;
142142         } else {
143143             return template.opsForValue().get(key);//根据key获取缓存中的val
144144         }
145145     }
146146 
147147 
148148 }

4） 消息发布和监听的服务类

1 1 package com.bigcustomer.utils.redisUtil;
2 2 
3 3 import com.bigcustomer.configs.BaseConfig;
4 4 import huashitech.kissucomponent.service.BaseService;
5 5 import org.springframework.beans.factory.annotation.Autowired;
6 6 import org.springframework.data.redis.core.StringRedisTemplate;
7 7 import org.springframework.stereotype.Service;
8 8 
9 9 /**
1010  * @author :CX
1111  * @Date :Create in 2018/8/23 10:22
1212  * @Effect :  redis 通道消息发送和监听接受
1313  */
1414 @Service
1515 public class RedisService extends BaseService {
1616 
1717     @Autowired
1818     private StringRedisTemplate template;
1919     @Autowired
2020     private BaseConfig baseConfig;
2121     @Autowired
2222     RedisService redisService;
2323 
2424     /**
2525      *@参数
2626      *@返回值
2727      *@创建人  cx
2828      *@创建时间
2929      *@描述       向默认通道发送消息
3030      */
3131     public void setMessage( Long funiqueid) {
3232 
3333         template.convertAndSend(baseConfig.getRedisAisle(),
3434                 baseConfig.getRedisMessageName() +funiqueid);
3535     }
3636 
3737 
3838     /**
3939      *@参数
4040      *@返回值
4141      *@创建人  cx
4242      *@创建时间
4343      *@描述       接受监听到的消息
4444      */
4545     public void receiveMessage(String message) {
4646         logger.info("接收redis通道消息："+message);
4747     }
4848 }

5） 使用

1 1 dao.getTransactionManager().doTransaction((TransactionStatus s) -> {
2 2                 //插入数据库
3 3                 int insert = dao.insert(tbCmChannelcustomerModel);
4 4                 // 加入缓存
5 5                 HashMap<String, Object> map = new HashMap<>();
6 6                 map.put(tbCmChannelcustomerModel.getFuniqueid().toString()
7 7                         , JSON.toJSONString(tbCmChannelcustomerModel));
8 8                 redisDao.setMap(map);
9 9                 // 发布redis通知消息
1010                 redisService.setMessage(tbCmChannelcustomerModel.getFuniqueid());
1111             });