当你做一个小型项目时，需要实现消息队列模式，但是却不想引用常见的MQ，因为它们实在是太重了，引入它们会给项目带来复杂度。而redis作为一个轻量级框架，主要作为缓存使用，而你正好可以通过redis实现消息的发布订阅。

一、如何使用

以SpringBoot项目为模板，来举例说明redis如何在Java中实现。

1.1 首先引入redis的jar包

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

1.2 配置redis连接池

@Configuration
public class JedisPoolFactory {
    @Bean
    public JedisPool redisFactory(){
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(10);
        jedisPoolConfig.setMaxWaitMillis(-1);
        jedisPoolConfig.setMinIdle(0);
        return new JedisPool(jedisPoolConfig, "127.0.0.1", 6379);
    }
}

1.3 发布消息

@Component
public class Publisher{

    @Autowired
    private JedisPool jedisPool;

    @Autowired
    private ObjectMapper objectMapper;
  	// 调用run方法传参，实现消息的发布
    public void run(StudentBean studentBean) {
        Jedis jedis = jedisPool.getResource();
        try {
            String value = objectMapper.writeValueAsString(studentBean);
            // 从mychannel的频道上推送消息
            jedis.publish("mychannel", value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}

1.4 消费者订阅

1. 订阅线程

@Slf4j
public class SubscribeThread extends Thread {

    private final JedisPool jedisPool;
    private final Subscriber subscriber = new Subscriber();

    private final String channel = "mychannel";

    public SubscribeThread(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    @Override
    public void run() {
        Jedis jedis = null;
        try {
            //从redis连接池中取出一个连接
            jedis = jedisPool.getResource();
            //通过subscribe 的api去订阅，入参是订阅者和频道名
            log.info("subscribe redis, channel: {}, thread will be blocked", channel);
            jedis.subscribe(subscriber, channel);
        } catch (Exception e) {
            log.error("subsrcibe channel error: {}", e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }
}

2. 业务处理，需要继承JedisPubSub，然后实现其发布订阅

public class Subscriber extends JedisPubSub {

    /**
     * 收到消息会调用
     * @param channel
     * @param message
     */
    @Override
    public void onMessage(String channel, String message) {
        // 收到message处理业务
        System.out.printf("received redis published message, channel: %s, message: %s%n",channel, message);
        // TODO message
    }

    /**
     * 订阅了频道会调用
     * @param channel
     * @param subscribedChannels
     */
    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        System.out.printf("subscribe redis channel success, channel %s, subscribedChannels %d%n",
                channel, subscribedChannels);
    }

    /**
     * 取消订阅会调用
     * @param channel
     * @param subscribedChannels
     */
    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) {
        System.out.printf("unsubscribe redis channel, channel %s, subscribedChannels %d%n",
                channel, subscribedChannels);
    }
}

二、总结

2.1 不能保证消息的可靠性

当发布一条消息时，如果没有订阅者或者订阅者断线时，那么这条消息就会被丢失，订阅者再次启动时无法获得该消息。

2.2 轻量级服务

如果在小型服务里面，只用到redis，而没有用到其它MQ中间件，那么可以用redis来实现消息队列，避免中间件的引入及安装。