Использование службы очереди сообщений Redis в Java

Java Redis

предисловие

Что касается Redis, мы уже обсуждали использование кеша, распределенной блокировки, распределенного уникального идентификатора и службы LBS, а здесь мы поговорим об использовании Redis для реализации службы сообщений.

Типичная служба сообщений — это служба производитель-потребитель. Как правило, производитель генерирует сообщение и отправляет его в очередь. Потребитель сообщения прослушивает сообщение и обрабатывает его.

Существует множество очень хороших продуктов для служб очередей сообщений. Например RabbitMQ, RocketMQ, Kafka и т.д. Эти продукты имеют очень продвинутые функции. Надежность и масштабируемость очень хорошие.

Но сам Redis также может легко реализовать шаблоны производителей и потребителей очередей сообщений. В этой статье кратко рассказывается, как это реализовано в Java.

Введение в связанные команды

Ссылаться наофициальный сайт редисКоманды, относящиеся к режимам redis и pubsub, следующие.

  • ПУБЛИКОВАТЬ сообщение канала: опубликовать сообщение в указанном канале.
  • ПОДПИСАТЬСЯ на канал [канал ...] : подписаться на сообщения из указанного канала. Каналов может быть несколько
  • PSUBSCRIBE pattern [шаблон ...] : Подпишитесь на сообщения, соответствующие заданному каналу шаблона.
  • UNSUBSCRIBE [канал [канал ...]] : прекратить подписку на сообщения с указанного канала.
  • PUNSUBSCRIBE [шаблон [шаблон ...]]: прекратить подписку на сообщения из указанного канала шаблона.
  • Подкоманда PUBSUB [аргумент [аргумент ...]]: это не ясно

Здесь мы в основном будем использовать первые три команда для реализации службы очереди сообщений на основе Redis.

Интерфейсы, связанные с джедаями

Отправить сообщение

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

запустить тест

Сначала запустите тест потребления сообщения, чтобы подписаться на сообщение, а затем запустите тестовый пример публикации сообщения, чтобы опубликовать сообщение. Результаты теста следующие

redis pubsub 运行结果

постскриптум

Эта процедура описана в предыдущей статье.«Использование Redis в Java»Это делается на основе добавления новых классов реализации. Код выпускается синхронно вРепозиторий GitHubсередина

Оригинальный текст был опубликован в брифе,оригинальная ссылка

Категории