redis实现异步消息处理

 

 如何使用redis实现消息的订阅与发布

 

 测试类

1. /**
2. *redis消息订阅发布测试类
3. */
4. @Controller
5. @RequestMapping(value="/queue")
6. publicclassTestControlller{
8. @Autowired
9. privateRedisTemplate redisService;
11. static{
12. newTestControlller().initThread("msgQueue");
13. }
16. /**订阅消息-启动消费队列数据线程**/
17. public void initThread(String key){
18. newThread(new blpopMessageHandler(key)).start();
19. }
22. /**发布消息--往队列里写数据**/
23. @RequestMapping(value="/brpush")
24. @UnSession
25. @ResponseBody
26. publicboolean brpush(String key,String name,int age)throwsException{
27. boolean flag = redisService.rpush(key,JSON.toJSONString(newEdu(name,age)));
28. return flag;
29. }
31. /**线程实时监听获取数据**/
32. class blpopMessageHandler implementsRunnable{
34. privateString key;
36. privateRedisTemplate redisTemplate=(RedisTemplate)SpringContextUtil.getBean("redisTemplate");
38. publicString getKey(){
39. return key;
40. }
42. publicvoid setKey(String key){
43. this.key = key;
44. }
46. public blpopMessageHandler(String key){
47. this.key = key;
48. }
50. @Override
51. publicvoid run(){
53. do{
55. //获取到数据
56. String result = redisTemplate.blpop(1000,key);
58. if(EmptyUtil.isNotEmpty(result)){
60. //do something
61. Edu edu = JSON.parseObject(result,Edu.class);
63. System.out.println("【线程一】"+edu.getName()+"---->"+edu.getAge());
64. }
66. }while(true);
68. }
69. }
72. }

model类

1.  
3. publicclassEduimplementsSerializable{
5. privatestaticfinallong serialVersionUID =-6336530413316596246L;
7. privateString name;
9. privateint age;
11. publicString getName(){
12. return name;
13. }
15. publicvoid setName(String name){
16. this.name = name;
17. }
19. publicint getAge(){
20. return age;
21. }
23. publicvoid setAge(int age){
24. this.age = age;
25. }
27. publicEdu(String name,int age){
28. this.name = name;
29. this.age = age;
30. }
32. @Override
33. publicString toString(){
34. return"Edu{"+
35. "name='"+ name +'\''+
36. ", age="+ age +
37. '}';
38. }
39. }

操作redis类

1.  
2. /**
3. * 实现Jedis对象的封装，实现操作
4. */
5. @Component
6. publicclassRedisTemplate{
7. privatestaticLogger logger =LoggerFactory.getLogger(RedisTemplate.class);
9. // 封装一个pool池用于jedis对象的管理
10. privatePool<Jedis> jedisPool;
12. publicRedisTemplate(Pool<Jedis> jedisPool){
13. this.jedisPool = jedisPool;
14. }
16. /**
17. * 存储REDIS队列 顺序存储
18. * @param key reids键名
19. * @param value 键值
20. */
21. publicboolean rpush(String key,String value){
23. Jedis jedis =null;
25. boolean flag =false;
26. try{
27. jedis = jedisPool.getResource();
28. jedis.rpush(key.getBytes("utf-8"),value.getBytes("utf-8"));
29. flag =true;
30. }catch(Exception e){
31. //释放redis对象
32. jedisPool.returnBrokenResource(jedis);
33. e.printStackTrace();
34. }finally{
35. //返还到连接池
36. close(jedis);
37. }
39. return flag;
40. }
42. /**
43. * 获取队列数据
44. * @param key 键名
45. * @return
46. */
47. publicString lpop(String key){
49. byte[] bytes =null;
50. Jedis jedis =null;
51. try{
53. jedis = jedisPool.getResource();
54. bytes = jedis.lpop(key.getBytes("utf-8"));
56. }catch(Exception e){
58. //释放redis对象
59. jedisPool.returnBrokenResource(jedis);
60. e.printStackTrace();
62. }finally{
64. //返还到连接池
65. close(jedis);
67. }
68. if(EmptyUtil.isNotEmpty(bytes)){
69. try{
70. returnnewString(bytes,"utf-8");
71. }catch(UnsupportedEncodingException e){
72. e.printStackTrace();
73. }
74. }
76. returnnull;
77. }
80. /**
81. * 获取阻塞list中的数据
82. * @param timeout
83. * @param key
84. * @return
85. */
86. publicString blpop(int timeout,String key){
88. List<String> list =newArrayList<>();
90. Jedis jedis =null;
92. try{
94. jedis = jedisPool.getResource();
96. list = jedis.blpop(timeout,key);
99. }catch(Exception e){
101. //释放redis对象
102. jedisPool.returnBrokenResource(jedis);
103. e.printStackTrace();
105. }finally{
107. //返还到连接池
108. close(jedis);
110. }
111. returnEmptyUtil.isNotEmpty(list)&& list.size()>1? list.get(1):null;
112. }
114. privatevoid close(Jedis jedis){
115. try{
116. jedisPool.returnResource(jedis);
117. }catch(Exception e){
118. if(jedis.isConnected()){
119. jedis.quit();
120. jedis.disconnect();
121. }
122. }
123. }
125. }