利用redis簡單實現訊息訂閱和釋出

阿新 • • 發佈：2022-03-27

利用redis簡單實現訊息訂閱和釋出
https://blog.csdn.net/Xs15073722800/article/details/63683750

好久沒寫部落格了,最近關於redis消費者,生產者的功能,看了很多的資料,個人覺得很多的MQ開源都很好用,redis相對小俏,簡單實現,一下先介紹redis實現;

1, redis實現訊息釋出和訂閱,

/**
 * Created by SDingBa.xiong on 17-3-9.
 */
public class RedisMsgPubSubListener extends JedisPubSub {
<span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">final</span> Logger logger = LoggerFactory.getLogger(RedisMsgPubSubListener.class);

<span class="hljs-annotation">@Resource</span>
<span class="hljs-keyword">private</span> AdOriginalityMapper adOriginalityMapper;

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">unsubscribe</span>() {
    <span class="hljs-keyword">super</span>.unsubscribe();
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">unsubscribe</span>(String... channels) {
    <span class="hljs-keyword">super</span>.unsubscribe(channels);
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">subscribe</span>(String... channels) {
    <span class="hljs-keyword">super</span>.subscribe(channels);
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">psubscribe</span>(String... patterns) {
    <span class="hljs-keyword">super</span>.psubscribe(patterns);
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">punsubscribe</span>() {
    <span class="hljs-keyword">super</span>.punsubscribe();
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">punsubscribe</span>(String... patterns) {
    <span class="hljs-keyword">super</span>.punsubscribe(patterns);
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onMessage</span>(String channel, String message) {
    logger.info(<span class="hljs-string">"channel={}, receives message={}"</span>, channel, message);


 code...
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onPMessage</span>(String pattern, String channel, String message) {

}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onSubscribe</span>(String channel, <span class="hljs-keyword">int</span> subscribedChannels) {
    logger.info(<span class="hljs-string">"====start Sub ====== channel={}, receives subscribedChannels={}"</span>, channel, subscribedChannels);
}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onPUnsubscribe</span>(String pattern, <span class="hljs-keyword">int</span> subscribedChannels) {

}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onPSubscribe</span>(String pattern, <span class="hljs-keyword">int</span> subscribedChannels) {

}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">onUnsubscribe</span>(String channel, <span class="hljs-keyword">int</span> subscribedChannels) {
    logger.info(<span class="hljs-string">"====end Sub ======= channel={}, receives subscribedChannels={}"</span>, channel, subscribedChannels);
}
 

}

2 spring boot初始化;

@Configuration
@Component
public class RedisMQSub implements ApplicationListener 
<ContextRefreshedEvent> {
<span class="hljs-keyword">private</span> <span class="hljs-keyword">final</span> Logger logger = LoggerFactory.getLogger(RedisMQSub.class);

@Resource
<span class="hljs-keyword">private</span> RedisService redisService;

<span class="hljs-comment">// 執行 監聽的 主機(123.56.86.1), 可以使用佇列解決 多個伺服器處理一個通知請求(先通過 MQ -&gt; 處理佇列)</span>
<span class="hljs-comment">// private static final String CRM_ACTIVE_HOSTNAME = "iZ25o9jw9hbZ";</span>
<span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">final</span> String CRM_ACTIVE_HOSTNAME = <span class="hljs-string">"su"</span>;

<span class="hljs-keyword">private</span> <span class="hljs-keyword">void</span> redisInit() {
    <span class="hljs-comment">// 本地測試 ;</span>
    <span class="hljs-comment">// Jedis jedis = new Jedis("localhost");</span>
    <span class="hljs-comment">// System.out.println((redisService == null) + " redis server");</span>
    <span class="hljs-comment">// RedisMsgPubSubListener listener = new RedisMsgPubSubListener();</span>
    <span class="hljs-comment">// new Thread(() -&gt; jedis.subscribe(listener, CreativeData.CREATIVE_ACTIVE_REDIS_MESSAGE_MQ)).start();</span>

    <span class="hljs-comment">// listener.unsubscribe();//關閉監聽</span>
}

@Override
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> onApplicationEvent(ContextRefreshedEvent event) {
    redisInit();
}
 

}

subscribe 會阻塞程序,,,,

jedis.public(通道名,訊息體…) 既可以釋出訊息

a,上面模式適合單一伺服器使用,,如果是分散式多伺服器,很多其他的MQ很好實現,redis需要使用到佇列功能了(因為訊息釋出和訂閱是一對多的模式(redis)),,,不能僅僅使用訊息的訂閱和釋出處理資料了,只能使用訊息的釋出用來通知其他的分散式節點,進行搶佔時向佇列裡面取資料,

e redis.lpush(); 和 redis.rpop() 可實現

o 工具類:

@Scope("singleton")
@Service
public class RedisService implements InitializingBean {
<span class="hljs-keyword">private</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">final</span> Logger LOGGER = LoggerFactory.getLogger(RedisService.class);
<span class="hljs-javadoc">/**
 * Redis操作介面
 *
 *<span class="hljs-javadoctag"> @author</span> 林計欽
 *<span class="hljs-javadoctag"> @version</span> 1.0 2013-6-14 上午08:54:14
 */</span>

<span class="hljs-annotation">@Resource</span>
<span class="hljs-keyword">private</span> SystemConfig systemConfig;
<span class="hljs-keyword">private</span> JedisPool pool = <span class="hljs-keyword">null</span>;

<span class="hljs-javadoc">/**
 * 返還到連線池
 *
 *<span class="hljs-javadoctag"> @param</span> redis
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title">returnResource</span>(Jedis redis) {
    <span class="hljs-keyword">if</span> (redis != <span class="hljs-keyword">null</span>) {
        redis.close();
    }
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">boolean</span> <span class="hljs-title">setNX</span>(String key, <span class="hljs-keyword">int</span> expired) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">int</span> result = <span class="hljs-number">0</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        result = jedis.setnx(key, <span class="hljs-string">"1"</span>).intValue();
        jedis.expire(key, expired);
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> result &gt; <span class="hljs-number">0</span>;
}

<span class="hljs-keyword">public</span> List&lt;String&gt; <span class="hljs-title">brpop</span>(String key) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    List&lt;String&gt; listStr = Lists.newArrayList();
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        listStr = jedis.brpop(<span class="hljs-number">10</span>, key);
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> listStr;
}

<span class="hljs-javadoc">/**
 * 獲取資料
 *
 *<span class="hljs-javadoctag"> @param</span> key
 *<span class="hljs-javadoctag"> @return</span>
 */</span>
<span class="hljs-keyword">public</span> String <span class="hljs-title">getString</span>(String key) {
    String value = <span class="hljs-keyword">null</span>;

    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        value = jedis.get(key);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        <span class="hljs-comment">// 釋放redis物件</span>
        <span class="hljs-keyword">if</span> (jedis != <span class="hljs-keyword">null</span>) {
            jedis.close();
        }
        LOGGER.error(<span class="hljs-string">"jedis_pool_err"</span>, e);
    } <span class="hljs-keyword">finally</span> {
        <span class="hljs-comment">// 返還到連線池</span>
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> value;
}

<span class="hljs-javadoc">/**
 * Strings結構：set command
 *
 *<span class="hljs-javadoctag"> @param</span> key
 *<span class="hljs-javadoctag"> @param</span> object 儲存object
 *<span class="hljs-javadoctag"> @param</span> expired 過期時間(秒)
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">boolean</span> <span class="hljs-title">setString</span>(String key, Object object, <span class="hljs-keyword">int</span> expired) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        jedis.setex(key, expired, String.valueOf(object));
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"redis error:key:{};value{}"</span>, key, object, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;
}

<span class="hljs-javadoc">/**
 * Hash結構：hset command
 *
 *<span class="hljs-javadoctag"> @param</span> key cache key
 *<span class="hljs-javadoctag"> @param</span> field second key
 *<span class="hljs-javadoctag"> @param</span> obj 直接存物件
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">hset</span>(String key, String field, Object obj, <span class="hljs-keyword">int</span> expired) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        jedis.hset(key, field, JSON.toJSONString(obj));
        jedis.expire(key.getBytes(), expired);
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }

}

<span class="hljs-annotation">@Override</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">afterPropertiesSet</span>() <span class="hljs-keyword">throws</span> Exception {
    JedisPoolConfig config = <span class="hljs-keyword">new</span> JedisPoolConfig();
    <span class="hljs-comment">// 控制一個pool可分配多少個jedis例項，通過pool.getResource()來獲取；</span>
    <span class="hljs-comment">// 如果賦值為-1，則表示不限制；如果pool已經分配了maxActive個jedis例項，則此時pool的狀態為exhausted(耗盡)。</span>
    config.setMaxTotal(<span class="hljs-number">500</span>);
    <span class="hljs-comment">// 控制一個pool最多有多少個狀態為idle(空閒的)的jedis例項。</span>
    config.setMaxIdle(<span class="hljs-number">5</span>);
    <span class="hljs-comment">// 表示當borrow(引入)一個jedis例項時，最大的等待時間，如果超過等待時間，則直接丟擲JedisConnectionException；</span>
    config.setMaxWaitMillis(<span class="hljs-number">1000</span> * <span class="hljs-number">100</span>);
    <span class="hljs-comment">// 在borrow一個jedis例項時，是否提前進行validate操作；如果為true，則得到的jedis例項均是可用的；</span>
    config.setTestOnBorrow(<span class="hljs-keyword">true</span>);
    String redisIp = systemConfig.getRedisIp();
    String redisPassword = systemConfig.getRedisPassword();
    <span class="hljs-keyword">int</span> redisPort = systemConfig.getRedisPort();
    pool = <span class="hljs-keyword">new</span> JedisPool(config, redisIp, redisPort, <span class="hljs-number">2000</span>, redisPassword);

}

<span class="hljs-javadoc">/**
 * 獲取資料
 *
 *<span class="hljs-javadoctag"> @param</span> key
 *<span class="hljs-javadoctag"> @return</span>
 */</span>
<span class="hljs-keyword">public</span> Map&lt;String, String&gt; <span class="hljs-title">hgetAll</span>(String key) {
    Map&lt;String, String&gt; value = <span class="hljs-keyword">null</span>;

    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        value = jedis.hgetAll(key);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        <span class="hljs-comment">// 釋放redis物件</span>
        <span class="hljs-keyword">if</span> (jedis != <span class="hljs-keyword">null</span>) {
            jedis.close();
        }
        LOGGER.error(<span class="hljs-string">"jedis_pool_err"</span>, e);
    } <span class="hljs-keyword">finally</span> {
        <span class="hljs-comment">// 返還到連線池</span>
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> value;
}

<span class="hljs-keyword">public</span> String <span class="hljs-title">hget</span>(String key, String field) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        <span class="hljs-keyword">return</span> jedis.hget(key, field);
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> <span class="hljs-string">""</span>;
}

<span class="hljs-keyword">public</span> Set&lt;String&gt; <span class="hljs-title">keys</span>(String field) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        <span class="hljs-keyword">return</span> jedis.hkeys(field);
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"redis_error.key:{}"</span>, field, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">null</span>;
}

<span class="hljs-javadoc">/**
 * Hash結構：hset command
 *
 *<span class="hljs-javadoctag"> @param</span> key cache key
 *<span class="hljs-javadoctag"> @param</span> field second key
 *<span class="hljs-javadoctag"> @param</span> obj 直接存物件
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">hset</span>(String key, String field, Object obj) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        jedis.hset(key, field, JSON.toJSONString(obj));
    } <span class="hljs-keyword">catch</span> (CacheException e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e.getMessage());
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }

}

<span class="hljs-javadoc">/**
 * 儲存REDIS佇列 順序儲存
 *
 *<span class="hljs-javadoctag"> @param</span> key reids鍵名
 *<span class="hljs-javadoctag"> @param</span> value 鍵值
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">lpush</span>(String key, String value) {

    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        jedis.lpush(key, value);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }

}

<span class="hljs-javadoc">/**
 * 獲取佇列資料
 *
 *<span class="hljs-javadoctag"> @param</span> key 鍵名
 */</span>
<span class="hljs-keyword">public</span> String <span class="hljs-title">rpop</span>(String key) {
    String bytes = <span class="hljs-keyword">null</span>;
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        bytes = jedis.rpop(key);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> bytes;
}

<span class="hljs-javadoc">/**
 * 獲取 佇列 長度
 *
 *<span class="hljs-javadoctag"> @param</span> key 佇列名
 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">long</span> <span class="hljs-title">llen</span>(String key) {
    <span class="hljs-keyword">long</span> bytes = <span class="hljs-number">0</span>;
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        bytes = jedis.llen(key);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"sedis_error.key:{}"</span>, key, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
    <span class="hljs-keyword">return</span> bytes;
}

<span class="hljs-javadoc">/**
 * 接收訊息。在main方法呼叫後，會一直執行下去。當有釋出對應訊息時，就會在jedisPubSub中接收到！
 *
 * 存在堵塞程序
 *<span class="hljs-javadoctag"> @param</span> jedisPubSub
 *<span class="hljs-javadoctag"> @param</span> channels

 */</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">subscribeMsg</span>(JedisPubSub jedisPubSub, String channels) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();
        jedis.subscribe(jedisPubSub, channels);
        LOGGER.debug(<span class="hljs-string">"subscribeMsg {} = {}"</span>, jedisPubSub, channels);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"subscribeMsg {} = {}"</span>, jedisPubSub, channels, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">publishMsg</span>(String channel, String message) {
    Jedis jedis = <span class="hljs-keyword">null</span>;
    <span class="hljs-keyword">try</span> {
        jedis = pool.getResource();

        jedis.publish(channel, message);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        LOGGER.error(<span class="hljs-string">"publishMsg {} = {}"</span>, channel, message, e);
    } <span class="hljs-keyword">finally</span> {
        returnResource(jedis);
    }
}

}