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

简介: 这里我使用Redis的发布、订阅功能实现简单的消息队列,基本的命令有publish、subscribe等。 在Jedis中,有对应的java方法,但是只能发布字符串消息。为了传输对象,需要将对象进行序列化,并封装成字符串进行处理。

这里我使用Redis的发布、订阅功能实现简单的消息队列,基本的命令有publish、subscribe等。

在Jedis中,有对应的java方法,但是只能发布字符串消息。为了传输对象,需要将对象进行序列化,并封装成字符串进行处理。


使用Redis实现消息队列

1.封装一个消息对象

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;
}
}

  

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

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();
 
}
}

  

3.从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;
}
}

  

4.创建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);
}
}

  

5.创建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);
}
}

  

6.测试方法

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();
}
}

  

 


相关实践学习
基于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
目录
相关文章
|
7天前
|
缓存 NoSQL Java
Java高并发实战:利用线程池和Redis实现高效数据入库
Java高并发实战:利用线程池和Redis实现高效数据入库
26 0
|
1天前
|
NoSQL Java Redis
如何在 Java 中操作这些 Redis 数据结构的基本方法
如何在 Java 中操作这些 Redis 数据结构的基本方法
8 2
|
7天前
|
NoSQL 测试技术 Scala
java.lang.RuntimeException: Can‘t start redis server. Check logs for details.
java.lang.RuntimeException: Can‘t start redis server. Check logs for details.
9 1
|
17天前
|
NoSQL Java Redis
【Redis】 Java操作客户端命令——列表操作与哈希操作
【Redis】 Java操作客户端命令——列表操作与哈希操作
|
17天前
|
NoSQL Java Linux
【Redis】 使用Java操作Redis的客户端
【Redis】 使用Java操作Redis的客户端
|
23天前
|
存储 消息中间件 缓存
Redis - 笔记
Redis是开源的内存数据结构存储系统,兼备数据库、缓存和消息中间件功能。它支持字符串、哈希、列表、集合、有序集合等数据结构,以及地理空间、基数统计和位图等特殊类型。Redis具备复制、LUA脚本、LRU事件、事务、持久化、哨兵和集群等高级特性,以实现高可用性。应用场景包括计数器、存储用户信息、消息队列、共同关注等。字符串最大容量为512M。
|
28天前
|
NoSQL 安全 Java
Java核心-redis
Java核心-redis
20 1
|
7天前
|
存储 消息中间件 NoSQL
使用 Java 操作 Redis 数据类型的详解指南
使用 Java 操作 Redis 数据类型的详解指南
7 0
|
30天前
|
缓存 NoSQL Redis
Java技术栈Redis面试总结(全面,实时更新)
Java技术栈Redis面试总结(全面,实时更新)
|
9天前
|
消息中间件 监控 Java
Java一分钟之-Kafka:分布式消息队列
【6月更文挑战第11天】Apache Kafka是一款高性能的消息队列,适用于大数据处理和实时流处理,以发布/订阅模型和分布式设计处理大规模数据流。本文介绍了Kafka基础,包括生产者、消费者、主题和代理,以及常见问题:分区选择、偏移量管理和监控不足。通过Java代码示例展示了如何创建生产者和消费者。理解并妥善处理这些问题,结合有效的监控和配置优化,是充分发挥Kafka潜力的关键。
15 0

热门文章

最新文章