Redis之Spring實現發佈訂閱

注：Redis版本是4.0；Spring版本4.3.11；Redis client版本2.9.0。

    首先開啓Redis服務。

1.建立ConnectionFactory和RedisTemplate，我用的是Fastjson的序列化

    List-1

import java.util.concurrent.TimeUnit;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @author mjduan@yahoo.com mjduan 2018-06-21 14:35
 * @version 1.0
 * @since 1.0
 */
@Configuration
public class RedisConfiguration {

    @Bean(name = "JedisConnectionFactory")
    public JedisConnectionFactory getJedisConnectionFactory() {
        JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
        jedisConnectionFactory.setPort(6379);
        jedisConnectionFactory.setHostName("localhost");
        return jedisConnectionFactory;
    }

    @Bean(name = "RedisTemplate")
    public RedisTemplate getRedisTemplate(@Qualifier(value = "JedisConnectionFactory") JedisConnectionFactory jedisConnectionFactory) {
        RedisTemplate redisTemplate = new RedisTemplate();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new CustomSerializer());
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    private static class CustomSerializer implements RedisSerializer {

        @Override
        public byte[] serialize(Object obj) throws SerializationException {
            return JSON.toJSONString(obj).getBytes();
        }

        @Override
        public Object deserialize(byte[] bytes) throws SerializationException {
            return null != bytes ? JSON.parseObject(bytes, JSONObject.class) : null;
        }
    }
}

2.建立MessageListener

    List-2

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.mjduan.project.util.fastjson.FastjsonSerialization;

/**
 * @author mjduan@yahoo.com mjduan 2018-06-24 14:39
 * @version 1.0
 * @since 1.0
 */
public class CustomMessageListener implements MessageListener {
    private static final StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
    private FastjsonSerialization fastjsonSerialization = FastjsonSerialization.getInstance();

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String channel = stringRedisSerializer.deserialize(message.getChannel());
        CustomMessage customMessage = fastjsonSerialization.deserialize(message.getBody(), CustomMessage.class);
        System.out.println("channel:" + channel + "; message:" + customMessage);
    }
}

3.配置ListenerContainer

    List-3 

import java.util.Arrays;
import java.util.List;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

/**
 * @author mjduan@yahoo.com mjduan 2018-06-24 14:51
 * @version 1.0
 * @since 1.0
 */
@Configuration
public class SubscribeConfiguration {

    @Bean(value = "MessageListener")
    public MessageListenerAdapter getMessageListenerAdapter() {
        return new MessageListenerAdapter(new CustomMessageListener());
    }

    @Bean(value = "RedisMessageListenerContainer")
    public RedisMessageListenerContainer getRedisMessageListenerContainer(
            @Qualifier(value = "JedisConnectionFactory") JedisConnectionFactory jedisConnectionFactory,
            @Qualifier(value = "MessageListener") MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(jedisConnectionFactory);
        List<ChannelTopic> topics = Arrays.asList(new ChannelTopic(PublishSubscribeTest.CHAT_ROOM),
                new ChannelTopic(PublishSubscribeTest.NEWS_CHANNEL));
        container.addMessageListener(messageListenerAdapter, topics);
        return container;
    }
}

4.定義一個通用的消息類

    List-4

/**
 * @author mjduan@yahoo.com mjduan 2018-06-24 14:42
 * @version 1.0
 * @since 1.0
 */
public class CustomMessage {
    private Type type;
    private Object content;

    public Type getType() {
        return type;
    }

    public void setType(Type type) {
        this.type = type;
    }

    public Object getContent() {
        return content;
    }

    public void setContent(Object content) {
        this.content = content;
    }

    enum Type{
        ORDER,PAYMENT,SHORT_MESSAGE
    }

    @Override
    public String toString() {
        return "CustomMessage{" +
                "type=" + type +
                ", content=" + content +
                '}';
    }
}

5.寫單元測試驗證

    List-5

import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.junit.Test;

import org.springframework.data.redis.core.RedisTemplate;

import com.mjduan.project.service.ServiceBaseTest;

/**
 * @author mjduan@yahoo.com mjduan 2018-06-24 14:32
 * @version 1.0
 * @since 1.0
 */
public class PublishSubscribeTest extends ServiceBaseTest {
    public static final String CHAT_ROOM = "chatRoom";
    public static final String NEWS_CHANNEL = "newsChannel";
    @Resource(name = "RedisTemplate")
    private RedisTemplate redisTemplate;

    @Test
    public void testSend() throws InterruptedException {
        CustomMessage customMessage = new CustomMessage();
        customMessage.setType(CustomMessage.Type.SHORT_MESSAGE);
        customMessage.setContent("Hi, 德洛麗絲");
        redisTemplate.convertAndSend(CHAT_ROOM, customMessage);

        customMessage.setContent("This is news");
        redisTemplate.convertAndSend(NEWS_CHANNEL, customMessage);

        TimeUnit.SECONDS.sleep(1);
    }
}

Reference:

1. Spring-redis文檔: https://docs.spring.io/spring-data/redis/docs/2.0.8.RELEASE/reference/html/

2.官網文檔給出的只是個簡單的例子，一些代碼片斷，咱們還要本身看源碼，瞭解還能夠怎麼使用，This is much more important.