redis 集群发布订阅 Java 实现

墨蓝 2022-05-25 21:05 245阅读 0赞

redis 集群 发布订阅 Java 实现。  
生产者发布消息到redis，redis订阅者自动弹出发布消息。  
发布者代码：

package cn.org.redis.demo;
    
    import redis.clients.jedis.JedisCluster;
    
    /*************************************************************************** * <pre></pre> * * @文件名称: ${FILE_NAME} * @包 路 径： cn.org.redis.demo * @版权所有： (C) 2018 * @类描述: * @版本: V2.0 * @创建人： july_whj * @创建时间：2018/4/25 16:39 ***************************************************************************/
    public class PushMsg { 
    
        public static final String CH = "CH_MESSAGE";
        //发布通道
        public final static String REDISCHANNEL = "ANYSIGN_SIGN_PDF";
        public static void main(String[] args) {
            JedisCluster jedisCluster = Pubsub.pubsubjava();
            jedisCluster.publish(REDISCHANNEL,"你好");
        }
    
    }

消息订阅者：

package cn.org.redis.demo;
    
    import redis.clients.jedis.HostAndPort;
    import redis.clients.jedis.JedisCluster;
    import redis.clients.jedis.JedisPoolConfig;
    
    import java.util.LinkedHashSet;
    import java.util.Set;
    
    /*************************************************************************** * <pre></pre> * * @文件名称: ${FILE_NAME} * @包 路 径： cn.org.redis.demo * @版权所有： (C) 2018 * @类描述: * @版本: V2.0 * @创建人： july_whj * @创建时间：2018/4/25 16:28 ***************************************************************************/
    public class Pubsub { 
    
        public static JedisCluster pubsubjava() {
            // TODO Auto-generated method stub
            JedisPoolConfig poolConfig = new JedisPoolConfig();
            // 最大连接数
            poolConfig.setMaxTotal(10);
            // 最大空闲数
            poolConfig.setMaxIdle(1);
            // 最大允许等待时间，如果超过这个时间还未获取到连接，则会报JedisException异常：
            // Could not get a resource from the pool
            poolConfig.setMaxWaitMillis(1000);
            Set<HostAndPort> nodes = new LinkedHashSet<HostAndPort>();
            nodes.add(new HostAndPort("192.168.136.136", 9001));
            nodes.add(new HostAndPort("192.168.136.136", 9002));
            nodes.add(new HostAndPort("192.168.136.136", 9003));
            nodes.add(new HostAndPort("192.168.136.136", 9004));
            nodes.add(new HostAndPort("192.168.136.136", 9005));
            nodes.add(new HostAndPort("192.168.136.136", 9006));
            return new JedisCluster(nodes, poolConfig);
    
        }
    
        public static void main(String[] args) {
            JedisCluster cluster = pubsubjava();
            RedisMQHandler sp = new RedisMQHandler();
            cluster.subscribe(sp,PushMsg.REDISCHANNEL);
        }
    
    
    }

package cn.org.redis.demo;
    
    import redis.clients.jedis.JedisPubSub;
    
    /*************************************************************************** * <pre></pre> * * @文件名称: ${FILE_NAME} * @包 路 径： cn.org.redis.demo * @版权所有： (C) 2018 * @类描述: * @版本: V2.0 * @创建人： july_whj * @创建时间：2018/4/25 16:23 ***************************************************************************/
    public class RedisMQHandler extends JedisPubSub { 
    
        /** * 监听到订阅模式接受到消息时的回调 (onPMessage) * 监听到订阅频道接受到消息时的回调 (onMessage ) * 订阅频道时的回调( onSubscribe ) * 取消订阅频道时的回调( onUnsubscribe ) * 订阅频道模式时的回调 ( onPSubscribe ) * 取消订阅模式时的回调( onPUnsubscribe ) * @param channel * @param message */
    
        @Override
        public void onMessage(String channel, String message) {
            // TODO Auto-generated method stub
            System.out.println(channel + "," + message);
        }
    
        @Override
        public void onPMessage(String pattern, String channel, String message) {
            // TODO Auto-generated method stub
            System.out.println(pattern + "," + channel + "," + message);
        }
    
        @Override
        public void onSubscribe(String channel, int subscribedChannels) {
            // TODO Auto-generated method stub
            System.out.println("onSubscribe: channel[" + channel + "]," + "subscribedChannels[" + subscribedChannels + "]");
        }
    
        @Override
        public void onUnsubscribe(String channel, int subscribedChannels) {
            // TODO Auto-generated method stub
            System.out.println(
                    "onUnsubscribe: channel[" + channel + "], " + "subscribedChannels[" + subscribedChannels + "]");
        }
    
        @Override
        public void onPUnsubscribe(String pattern, int subscribedChannels) {
            // TODO Auto-generated method stub
            System.out.println("onPUnsubscribe: pattern[" + pattern + "]," +
                    "subscribedChannels[" + subscribedChannels + "]");
    
        }
    
        @Override
        public void onPSubscribe(String pattern, int subscribedChannels) {
            System.out.println("onPSubscribe: pattern[" + pattern + "], " +
                    "subscribedChannels[" + subscribedChannels + "]");
    
        }
    
    }

接收发布者消息：

ANYSIGN_SIGN_PDF,{}
    ANYSIGN_SIGN_PDF,{}
    ANYSIGN_SIGN_PDF,{
       "statusCode":"10304"}
    ANYSIGN_SIGN_PDF,{
       "msg":"{\"messageBodyReferences\":[{\"fileExtType\":\"PDF\",\"num\":1,\"fileSize\":173956}]}","pfdFilePaths":["E:\\huataizhengjuan\\out\\testAnySignEncPackage_ba340e82-fe72-4fcd-8e72-1f40efdf82b6-0.pdf"],"statusCode":"200","taskId":"ba340e82-fe72-4fcd-8e72-1f40efdf82b6"}