Redis笔记(七)Java实现Redis消息队列

简介:

封装一个消息对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public  class  Message  implements  Serializable{
 
private  static  final  long  serialVersionUID = 1L;
 
private  String titile;
private  String info;
 
public  Message(String titile,String info){
this .titile=titile;
this .info=info;
}
 
public  String getTitile() {
return  titile;
}
public  void  setTitile(String titile) {
this .titile = titile;
}
public  String getInfo() {
return  info;
}
public  void  setInfo(String info) {
this .info = info;
}
}

  

为这个消息对象提供序列化方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public  class  MessageUtil {
 
//convert To String
public  static  String convertToString(Object obj,String charset)  throws  IOException{
 
ByteArrayOutputStream bo =  new  ByteArrayOutputStream();
ObjectOutputStream oo =  new  ObjectOutputStream(bo);
oo.writeObject(obj);
String str = bo.toString(charset);
bo.close();
oo.close();
return  str;
}
 
//convert To Message
public  static  Object convertToMessage( byte [] bytes)  throws  Exception{
ByteArrayInputStream in =  new  ByteArrayInputStream(bytes);
ObjectInputStream sIn =  new  ObjectInputStream(in);
return  sIn.readObject();
 
}
}

  

从Jedis连接池中获取连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public  class  RedisUtil {
 
/**
* Jedis connection pool
* @Title: config
*/
public  static  JedisPool getJedisPool(){
ResourceBundle bundle=ResourceBundle.getBundle( "redis" );
String host=bundle.getString( "host" );
int  port=Integer.valueOf(bundle.getString( "port" ));
int  timeout=Integer.valueOf(bundle.getString( "timeout" ));
//  String password=bundle.getString("password");
 
JedisPoolConfig config= new  JedisPoolConfig();
config.setMaxActive(Integer.valueOf(bundle.getString( "maxActive" )));
config.setMaxWait(Integer.valueOf(bundle.getString( "maxWait" )));
config.setTestOnBorrow(Boolean.valueOf(bundle.getString( "testOnBorrow" )));
config.setTestOnReturn(Boolean.valueOf(bundle.getString( "testOnReturn" )));
 
JedisPool pool= new  JedisPool(config, host, port, timeout);
 
return  pool;
}
}

  

创建Provider类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public  class  Producer {
 
private  Jedis jedis;
private  JedisPool pool;
 
public  Producer(){
pool=RedisUtil.getJedisPool();
jedis = pool.getResource();
}
 
 
public  void  provide(String channel,Message message)  throws  IOException{
String str1=MessageUtil.convertToString(channel, "UTF-8" );
String str2=MessageUtil.convertToString(message, "UTF-8" );
jedis.publish(str1, str2);
}
 
//close the connection
public  void  close()  throws  IOException {
//将Jedis对象归还给连接池,关闭连接
pool.returnResource(jedis);
}
}

  

创建Consumer类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
public  class  Consumer {
 
private  Jedis jedis;
private  JedisPool pool;
 
public  Consumer(){
pool=RedisUtil.getJedisPool();
jedis = pool.getResource();
}
 
 
public  void  consum(String channel)  throws  IOException{
JedisPubSub jedisPubSub =  new  JedisPubSub() {
// 取得订阅的消息后的处理
public  void  onMessage(String channel, String message) {
System.out.println( "Channel:" +channel);
System.out.println( "Message:" +message.toString());
}
 
// 初始化订阅时候的处理
public  void  onSubscribe(String channel,  int  subscribedChannels) {
System.out.println( "onSubscribe:" +channel);
}
 
// 取消订阅时候的处理
public  void  onUnsubscribe(String channel,  int  subscribedChannels) {
System.out.println( "onUnsubscribe:" +channel);
}
 
// 初始化按表达式的方式订阅时候的处理
public  void  onPSubscribe(String pattern,  int  subscribedChannels) {
// System.out.println(pattern + "=" + subscribedChannels);
}
 
// 取消按表达式的方式订阅时候的处理
public  void  onPUnsubscribe(String pattern,  int  subscribedChannels) {
// System.out.println(pattern + "=" + subscribedChannels);
}
 
// 取得按表达式的方式订阅的消息后的处理
public  void  onPMessage(String pattern, String channel, String message) {
System.out.println(pattern +  "="  + channel +  "="  + message);
}
};
 
jedis.subscribe(jedisPubSub, channel);
}
 
//close the connection
public  void  close()  throws  IOException {
//将Jedis对象归还给连接池
pool.returnResource(jedis);
}
}

  

测试方法

1
2
3
4
5
6
7
8
9
10
11
12
13
public  static  void  main(String[] args){
 
Message msg= new  Message( "hello!" "this is the first message!" );
 
Producer producer= new  Producer();
Consumer consumer= new  Consumer();
try  {
producer.provide( "chn1" ,msg);
consumer.consum( "chn1" );
catch  (IOException e) {
e.printStackTrace();
}
}

  


本文转自邴越博客园博客,原文链接:http://www.cnblogs.com/binyue/p/4763352.html,如需转载请自行联系原作者

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
8天前
|
JSON NoSQL Java
Redis入门到通关之Java客户端SpringDataRedis(RedisTemplate)
Redis入门到通关之Java客户端SpringDataRedis(RedisTemplate)
215 0
|
8天前
|
canal 缓存 NoSQL
【Redis系列笔记】双写一致性
本文讨论了缓存不一致问题及其后果,如价格显示错误和订单计算错误。问题主要源于并发和双写操作的异常。解决方案包括使用分布式锁(但可能导致性能下降和复杂性增加)、延迟双删策略(通过延迟删除缓存来等待数据同步)以及异步同步方法,如通过Canal和MQ实现数据的最终一致性。面试中,可以提及这些策略来确保数据库和缓存数据的一致性。
116 1
【Redis系列笔记】双写一致性
|
8天前
|
缓存 NoSQL 安全
【Redis系列笔记】缓存三剑客
缓存穿透是指请求一个不存在的数据,缓存层和数据库层都没有这个数据,这种请求会穿透缓存直接到数据库进行查询。它通常发生在一些恶意用户可能故意发起不存在的请求,试图让系统陷入这种情况,以耗尽数据库连接资源或者造成性能问题。 缓存击穿发生在访问热点数据,大量请求访问同一个热点数据,当热点数据失效后同时去请求数据库,瞬间耗尽数据库资源,导致数据库无法使用。 缓存雪崩是缓存中大量key失效后当高并发到来时导致大量请求到数据库,瞬间耗尽数据库资源,导致数据库无法使用。
52 1
|
8天前
|
NoSQL Java 关系型数据库
【Redis系列笔记】分布式锁
分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。 分布式锁的核心思想就是让大家都使用同一把锁,只要大家使用的是同一把锁,那么我们就能锁住线程,不让线程进行,让程序串行执行,这就是分布式锁的核心思路
138 2
|
1天前
|
缓存 NoSQL Redis
Java技术栈Redis面试总结(全面,实时更新)
Java技术栈Redis面试总结(全面,实时更新)
|
3天前
|
NoSQL Java Redis
如何在redis中设置文件夹名java - 蓝易云
在这个例子中,"folder:subfolder:key"就像一个文件夹路径,并且键的值被设置为"value"。然后,我们使用相同的键来获取并打印值。最后,记得关闭Jedis连接。
33 0
|
6天前
|
NoSQL 算法 Java
【redis源码学习】持久化机制,java程序员面试算法宝典pdf
【redis源码学习】持久化机制,java程序员面试算法宝典pdf
|
8天前
|
NoSQL Java Redis
在Java中操作Redis
在Java中操作Redis
15 0
|
8天前
|
存储 NoSQL Redis
【Redis系列笔记】Redis总结
Redis是一个基于内存的 key-value 结构数据库。 Redis 是互联网技术领域使用最为广泛的存储中间件。 Redis是用C语言开发的一个开源的高性能键值对(key-value)数据库,官方提供的数据是可以达到100000+的QPS(每秒内查询次数)。 它存储的value类型比较丰富,也被称为结构化的NoSql数据库。
68 0
|
8天前
|
缓存 NoSQL Java
【Redis系列笔记】Redis入门
本文介绍了Redis常用命令,以及SpringBoot集成Spring Data Redis和Spring Cache。Spring Data Redis 提供了对 Redis 的操作方法,而 Spring Cache 则提供了基于注解的缓存功能,可以方便地将方法的返回值缓存到 Redis 中,以提高性能和减少对数据源的访问次数。这样的集成可以帮助开发者更便捷地利用 Redis 来管理应用程序的数据和缓存。
101 4

热门文章

最新文章