Java实现Redis发布/订阅

时间 2019-11-07

标签 java 实现 redis 发布订阅栏目 Java 繁體版

原文原文链接

今天经理让我实现一个Redis发布/订阅功能，用来记录审计信息。我查了一天，才弄出来。总结以下，大神勿喷。次日我百度“JedisPubSub ”发现好多代码，后来测试了一下发现，百度关键词“Jedis订阅发布”确实没多少能够参考的，可是百度关键词“Jedis的Publish/Subscribe”代码就有不少，我也是醉了。html

redis的发布/订阅模式是消息机制之一，另一个叫生成者消费者模式。Redis发布订阅模式讲解能够参考菜鸟教程的这篇文章http://www.runoob.com/redis/redis-pub-sub.html。java

一、Redis发布订阅模式客户端实现。在打开Redis服务器后，再打开两个客户端，客户端1用来接收消息，客户端2用来发布消息。mysql

客户端1订阅bar频道。格式：SUBSCRIBE name1 name2。
成功订阅回复，分别对应订阅类型、订阅频道、订阅数量。web

127.0.0.1:6379> SUBSCRIBE bar
Reading messages... (press Ctrl-C to quit)1) "subscribe"2) "bar"3) (integer) 1

客户端2，发送消息。格式：publish channelName Message。redis

127.0.0.1:6379> publish bar val
(integer) 1

客户端1订阅回复，分别对应消息类型，频道，消息。spring

1) "message"2) "bar"3) "val"

效果图以下：sql

Redis支持模式匹配订阅，*为模糊匹配符。服务器

订阅全部频道的消息ide

PSUBSCRIBE *

订阅以news.开头的全部频道。工具

PSUBSCRIBE news.*

其余操做这里不在赘述。

二、Java实现Redis的发布订阅。

Java实现Redis的功能大部分是使用jedis的jar包来进行操做。我的感受，jedis封装了操做redis的经常使用命令，写多了就会发现知道redis命令怎么写的，就能够猜出来jedis中怎么写的。

首先在咱们封装的JedisUtils中加入发布和订阅操做的方法。你们没有JedisUtils的能够参考JeeSite中的JedisUtils怎么写的。

 /**
  * 发布一个消息
  * @param channel
  * @param message
  */
 public static void publishMsg(String channel,String message){
  Jedis jedis = null;
  try {
   jedis = getResource();
   
jedis.publish(channel, message);
   logger.debug("publishMsg {} = {}", channel, message);
  } catch (Exception e) {
   logger.warn("publishMsg {} = {}", channel, message, e);
  } finally {
   returnResource(jedis);
  }
 }
 
 /**
  * 发布一个消息
  * @param channel
  * @param message
  */
 public static void publishMsg(byte[] channel,byte[] message){
  Jedis jedis = null;
  try {
   jedis = getResource();
  
 jedis.publish(channel, message);
   logger.debug("publishMsg {} = {}", channel, message);
  } catch (Exception e) {
   logger.warn("publishMsg {} = {}", channel, message, e);
  } finally {
   returnResource(jedis);
  }
 }

上面的两个方法的核心处理就是jedis.publish(channel, message);参数channel是消息的频道，message是消息的内容。在Junit测试或者其余的地方，使用工具类的此方法便可发布一个消息。

接收消息代码多一些。首先定义一个类继承JedisPubSub，而后实现其中的未实现的方法，最后在工具类JedisUtils中定义一个操做的方法便可。代码以下：

 /**
  * 接收消息。在main方法调用后，会一直执行下去。当有发布对应消息时，就会在jedisPubSub中接收到！
  * @param jedisPubSub
  * @param channels
  */
 public static void subscribeMsg(JedisPubSub jedisPubSub,String channels){
  Jedis jedis = null;
  try {
   jedis = getResource();
   jedis.subscribe(jedisPubSub, channels);
   logger.debug("subscribeMsg {} = {}", jedisPubSub, channels);
  } catch (Exception e) {
   logger.warn("subscribeMsg {} = {}", jedisPubSub, channels, e);
  } finally {
   returnResource(jedis);
  }
 }

JedisPubSub类的定义以下：

package com.aq.web.shiro.redis.msg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisPubSub;
import com.aq.dao.info.RedisMsgAuditInfo;
import com.aq.web.shiro.redis.SelfObjectUtils;
/**
 * Redis监听订阅事件 
 * @author 
 *
 */
public class RedisMsgPubSubListener extends JedisPubSub{
 private static Logger logger = LoggerFactory.getLogger(RedisMsgPubSubListener.class);
 
 
 /**
  * 取得订阅的消息后的处理
  */
 @Override
 public void onMessage(String channel, String message) {
  logger.debug("onMessage: channel["+channel+"], message["+message+"]");
  
  //若是消息类型与定义的审计专用类型一致
  if(RedisMsgJedisUtils.getRedisMsgChannelString().equalsIgnoreCase(channel)){
   //处理审计的消息
   this.AuditMsgHandler(message);
  }
 }
 /**
  * 取得按表达式的方式订阅的消息后的处理
  */
 @Override
 public void onPMessage(String pattern, String channel, String message) {
  logger.debug("onPMessage: channel["+channel+"], message["+message+"]");
  
 }
 /**
  * 初始化订阅时候的处理   
  */
 @Override
 public void onSubscribe(String channel, int subscribedChannels) {
   logger.debug("onSubscribe: channel["+channel+"],"+
     "subscribedChannels["+subscribedChannels+"]");
  
 }
 /**
  * 取消订阅时候的处理 
  */
 @Override
 public void onUnsubscribe(String channel, int subscribedChannels) {
  logger.debug("onUnsubscribe: channel["+channel+"], "+
                                               "subscribedChannels["+subscribedChannels+"]");
  
 }
 /**
  * 取消按表达式的方式订阅时候的处理
  */
 @Override
 public void onPUnsubscribe(String pattern, int subscribedChannels) {
   logger.debug("onPUnsubscribe: pattern["+pattern+"],"+
                                               "subscribedChannels["+subscribedChannels+"]");
  
 }
 
 /**
  * 初始化按表达式的方式订阅时候的处理 
  */
 @Override
 public void onPSubscribe(String pattern, int subscribedChannels) {
  logger.debug("onPSubscribe: pattern["+pattern+"], "+
                                               "subscribedChannels["+subscribedChannels+"]");
  
 }
 
 /**
  * 私有方法：用于处理审计的消息
  */
 private void AuditMsgHandler(String message){
  //审计日志反序列化 String -> byte[] -> RedisMsgAuditInfo
  RedisMsgAuditInfo msg3 = (RedisMsgAuditInfo) SelfObjectUtils.unserialize(message.getBytes());
  logger.debug(msg3.toString());
  //TODO 后续怎么存先不写...
 }
 
}

审计日志反序列化部分代码能够省略掉。

‍

最后Junit中测试以下：

package ap.shiro.redis.msg;
import org.junit.Test;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
import com.aq.web.shiro.redis.JedisUtils;
import com.aq.web.shiro.redis.msg.RedisMsgJedisUtils;
import com.aq.web.shiro.redis.msg.RedisMsgPubSubListener;
@ContextConfiguration(locations = "/mysql-test.xml")
public class RedisMsgPubSubTester extends AbstractJUnit4SpringContextTests{
 
 @Test
 public void testMsg_pub(){
  
  RedisMsgJedisUtils.publishMsg("news.share", "2016年3月28日 15:34:37");
  
 }
 
 @Test
 public void testMsg_sub(){
  RedisMsgPubSubListener pubsub = new RedisMsgPubSubListener();
  RedisMsgJedisUtils.subscribeMsg(pubsub, "news.share");
  
 }
}

建议测试时，分开方法进行测试。其中testMsg_pub()方法是用来测试发布消息的,testMsg_sub()是用来测试接收订阅消息的。这里只是使用了普通订阅，你们还可使用模式订阅。执行testMsg_sub()方法后，客户端会一直开启着，不会关闭。另外，在其余的redis客户端中发布一条消息，控制台就会马上输出该消息。

这样Redis的订阅发布的Java实现就完成了。整体不太难，我今天搜资料的时候发现资料不少也很乱，固然没在开源中国搜。

另外，上面例子中使用的JedisPubSub类的子类接收消息。次日测试时发现，JedisPubSub类的子类接收字符串类的消息没问题，可是接收对象转byte[]的消息后不能正确地转换回对象。这里的对象是自定义的，经过ByteArray流和Object流完成Object与byte[]之间的转换。

通过查资料后发现，使用BinaryJedisPubSub类的子类接收消息能够正确地转换对象，不会出现上述问题。你们本身试下。继承BinaryJedisPubSub类的监听器跟JedisPubSub类的监听器相似。