在 Java 中使用 redis 的消息隊列服務

前言

關於 redis 咱們前面已經討論過了緩存、分佈式鎖、分佈式惟一標識、LBS服務的用法，這裏咱們來談談利用 redis 來實現一個消息服務。

典型的消息服務是一個生產者和消費者模式的服務。通常是有生產者產生消息，將消息發送到隊列中。而消息的消費者則監聽消息，對消息進行處理。

有不少很是優秀的消息隊列服務的產品。例如 RabbitMQ、RocketMQ、Kafka 等。這些產品都具有很是高級的功能。可靠性、擴展性都很是的好。

可是 redis 自身也可以很簡單的實現消息隊列的生產者和消費者模式。本文簡單介紹一下在 Java 下是如何實現的。

相關命令介紹

參考 redis 官網 redis 和 pubsub 模式相關的命令以下。

• PUBLISH channel message ：發佈一條消息到指定的頻道
• SUBSCRIBE channel [channel ...] ：訂閱指定頻道的消息。頻道能夠是多個
• PSUBSCRIBE pattern [pattern ...] ：訂閱符合給定模式頻道的消息
• UNSUBSCRIBE [channel [channel ...]] ：中止訂閱指定頻道的消息
• PUNSUBSCRIBE [pattern [pattern ...]] ：中止訂閱指定模式頻道的消息
• PUBSUB subcommand [argument [argument ...]] ： 這個沒搞清

這裏面咱們主要會用到前三個命令來實現一個基於 redis 的消息隊列服務。

Jedis 相關接口

發送消息

public Long publish(final String channel, final String message)
複製代碼

在這裏，咱們會用這個接口來向消息隊列指定的頻道發送消息，在 Java裏面的核心代碼以下：

jedis = redisConnection.getJedis();
jedis.publish(channel, message);
複製代碼

訂閱消息

public void subscribe(final JedisPubSub jedisPubSub, final String... channels)
複製代碼

在這裏，咱們用這個接口來訂閱指定頻道的消息，而且經過實現的 JedisPubSub 對象的回調接口來處理收到的消息。改邏輯核心代碼以下所示

Jedis jedis = redisConnection.getJedis();
jedis.subscribe(new JedisPubSub() {
    @Override
    public void onMessage(String channel, String message) {
        //處理消息邏輯代碼
     }
複製代碼

}, channels);

消息發佈完整實現

定義接口

package com.x9710.common.redis;

/**
 * 消息生產者服務接口
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public interface MessagePublisher {
    boolean sendMessage(String message);
}
複製代碼

實現接口

package com.x9710.common.redis.impl;

import com.x9710.common.redis.MessagePublisher;
import com.x9710.common.redis.RedisConnection;
import redis.clients.jedis.Jedis;

/**
 * 消息生產者服務 redis 實現類
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public class MessagePublisherRedisImpl implements MessagePublisher {
private RedisConnection redisConnection;

/**
 * 能夠同時向多個頻道發送數據
 */
private String[] channels;

public void setRedisConnection(RedisConnection redisConnection) {
    this.redisConnection = redisConnection;
}

public void setChannels(String[] channels) {
    this.channels = channels;
}

public boolean sendMessage(String message) {
    Jedis jedis = null;
    try {
        if (channels != null && channels.length > 0) {
            jedis = redisConnection.getJedis();
            for (String channel : channels) {
                jedis.publish(channel, message);
            }
            return true;
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (jedis != null) {
            jedis.close();
        }
    }
    return false;
}
}
複製代碼

消息消費完整實現

定義接口

package com.x9710.common.redis;

/**
 * 消息消費者服務接口
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public interface MessageConsumer {
    void handleMessage(String message);
}
複製代碼

實現接口

package com.x9710.common.redis.impl;

import com.x9710.common.redis.MessageConsumer;
import com.x9710.common.redis.RedisConnection;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * 消息消費者服務 redis 實現類
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public abstract class MessageConsumerRedisImpl implements MessageConsumer {

/**
 *
 * @param redisConnection redis 鏈接類
 * @param channels 訂閱的頻道列表
 */
public MessageConsumerRedisImpl(RedisConnection redisConnection, String[] channels) {
    Jedis jedis = null;
    try {
        if (channels != null && channels.length > 0) {
            jedis = redisConnection.getJedis();
            jedis.subscribe(new JedisPubSub() {
                @Override
                public void onMessage(String channel, String message) {
                    System.out.println("receive " + message + " from " + channel);
                    handleMessage(message);
                }
            }, channels);
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (jedis != null) {
            jedis.close();
        }
    }
}
}
複製代碼

這裏定義了一個抽象類，目的是讓用戶繼承該類來實現 handleMessage 方法，在該方法中實現收到消息後的業務邏輯。

測試

實現消息訂閱測試

package com.x9710.common.redis.test;

import com.x9710.common.redis.RedisConnection;
import com.x9710.common.redis.impl.MessageConsumerRedisImpl;
import org.junit.Test;

/**
 * 消息消費服務測試類
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public class RedisConsumerTest {

@Test
public void consumerTest() {
    RedisConnection redisConnection = RedisConnectionUtil.create();
    new MessageConsumerRedisImpl(redisConnection, new String[]{"channel1", "channel2"}) {
        public void handleMessage(String message) {
            System.out.println(message);
        }
    };
}

}
複製代碼

實現消息發送測試

package com.x9710.common.redis.test;

import com.x9710.common.redis.RedisConnection;
import com.x9710.common.redis.impl.MessagePublisherRedisImpl;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

/**
 * 消息發送服務測試類
 *
 * @author 楊高超
 * @since 2018-01-03
 */
public class RedisPublisherTest {
    private MessagePublisherRedisImpl messagePublisherRedis;

@Before
public void before() {
    RedisConnection redisConnection = RedisConnectionUtil.create();
    messagePublisherRedis = new MessagePublisherRedisImpl();
    messagePublisherRedis.setRedisConnection(redisConnection);
    messagePublisherRedis.setChannels(new String[]{"channel1", "channel2"});
}

@Test
public void publisherTest() {
    for (int i = 0; i < 3; i++) {
        Assert.assertTrue(messagePublisherRedis.sendMessage("message" + i));
    }
}
}
複製代碼

運行測試

首先運行消息消費測試用來訂閱消息，而後運行消息發佈測試用例來發布消息。測試結果以下

後記

這個程序實在前面的文章《在 Java 中使用 redis》的基礎上添加新的實現類的方式完成的。代碼同步發佈在 GitHub 倉庫中

原文發表在簡書中，原始連接