【Springboot+Redis】Springboot+Redis实现消息队列（生产者/消费者、发布订阅模式）

 

一般来说，消息队列有两种场景，一种是发布者订阅者模式，一种是生产者消费者模式。利用redis这两种场景的消息队列都能够实现。

定义：
        生产者消费者模式：生产者生产消息放到队列里，多个消费者同时监听队列，谁先抢到消息谁就会从队列中取走消息；即对于每个消息只能被最多一个消费者拥有。
        发布者订阅者模式：发布者生产消息放到队列里，多个监听队列的消费者都会收到同一份消息；即正常情况下每个消费者收到的消息应该都是一样的。

Redis不仅可作为缓存服务器，还可用作消息队列。它的列表类型天生支持用作消息队列。如下图所示：

由于Redis的列表是使用双向链表实现的，保存了头尾节点，所以在列表头尾两边插取元素都是非常快的。

在Redis中，List类型是按照插入顺序排序的字符串链表。和数据结构中的普通链表一样，我们可以在其头部(left)和尾部(right)添加新的元素。在插入时，如果该键并不存在，Redis将为该键创建一个新的链表。与此相反，如果链表中所有的元素均被移除，那么该键也将会被从数据库中删除。List中可以包含的最大元素数量是4294967295。
从元素插入和删除的效率视角来看，如果我们是在链表的两头插入或删除元素，这将会是非常高效的操作，即使链表中已经存储了百万条记录，该操作也可以在常量时间内完成。然而需要说明的是，如果元素插入或删除操作是作用于链表中间，那将会是非常低效的。相信对于有良好数据结构基础的开发者而言，这一点并不难理解。

Redis List的主要操作为lpush/lpop/rpush/rpop四种，分别代表从头部和尾部的push/pop，除此之外List还提供了两种pop操作的阻塞版本blpop/brpop，用于阻塞获取一个对象。

Redis通常都被用做一个处理各种后台工作或消息任务的消息服务器。 一个简单的队列模式就是：生产者把消息放入一个列表中，等待消息的消费者用 RPOP 命令（用轮询方式）， 或者用 BRPOP 命令（如果客户端使用阻塞操作会更好）来得到这个消息。

1、Redis 生产者/消费者模式实现消息队列的简单例子(此外用的是jedisCluster连接redis)

springboot的application-dev.yml中redis的配置如下：

redis:
    database: 0
    password:  # Redis服务器连接密码（默认为空）
    timeout: 0 # 连接超时时间（毫秒）
    commandTimeout: 5000
    cluster:
        nodes: 10.201.1.27:7001,10.201.1.27:7002,10.201.1.27:7003,10.201.1.27:7004,10.201.1.27:7005,10.201.1.27:7006

redis放入Spring容器中：

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import java.util.HashSet;
import java.util.Set;

@Configuration
public class RedisHostsConfig {

    @Value("${spring.redis.cluster.nodes}")
    private String hosts;

    /**
     * 获取jedis链接
     * @return
     */
    @Bean
    public JedisCluster redisCluster(){
    try{
        Set<HostAndPort> nodes = new HashSet<>();
        String[] hostArray = hosts.split(",");
        for(String host : hostArray){
            String[] ipAndPort = host.split(":");
            nodes.add(new HostAndPort(ipAndPort[0], Integer.parseInt(ipAndPort[1])));
        }
        JedisCluster cluster = new JedisCluster(nodes);
        return cluster;
    }catch(Exception e){
        e.printStackTrace();
        return null;
    }
    }
}

 

生产者代码：

public class RedisProducer {

    @Autowired
    JedisCluster jedisCluster;

    public static void main(String[] args){
        for(int i = 0;i<10;i++) {
            jedisCluster.lpush("listingList","value_" + i);  
        }
    }
 
}

 

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.JedisCluster;
import java.util.List;

@Slf4j
public class NetListingConsumer extends Thread {

    private static JedisCluster redisCluster;

    private static JedisCluster getRedisCluster(){
        if (redisCluster == null){
            redisCluster = SpringContextHolder.getBean("redisCluster");
        }
        return redisCluster;
    }

    @Override
    public void run() {
        JedisCluster redis = getRedisCluster();
        while (true){
            //阻塞式brpop，List中无数据时阻塞，参数0表示一直阻塞下去，直到List出现数据
            List<String> listingList = redis.brpop(0, "listingList");
            log.info("线程取数据：{}", listingList.get(1));
        }
    }
}

 

消费者代码：

运行消费者：

public class RedisProducer {
    public static void main(String[] args){
        PriceConsumer redisConsumer = new PriceConsumer();
        redisConsumer.start();
    }
}

spring-redis使用RedisMessageListenerContainer进行消息监听，客户程序需要自己实现MessageListener，并以指定的topic注册到RedisMessageListenerContainer，这样，在指定的topic上如果有消息，RedisMessageListenerContainer便会通知该MessageListener。2、Redis 发布/订阅模式实现消息队列的简单例子（此外用的是redisTemplater操作redis）

在此，我是监听了四个topic，为每一个topic写了一个消息处理方法。

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import java.util.concurrent.CountDownLatch;

@Configuration
public class RedisMessageListener {

    /**
     * redis消息监听器容器
     * @param connectionFactory
     * @param priceListenerAdapter  价格趋势和市场消息订阅处理器
     * @param listingListenerAdapter 挂牌案例消息订阅处理器
     * @param caseListenerAdapter  交易案例消息订阅处理器
     * @param foreclosureListenerAdapter  法拍案例消息订阅处理器
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                MessageListenerAdapter priceListenerAdapter,
                MessageListenerAdapter listingListenerAdapter,
                MessageListenerAdapter caseListenerAdapter,
                MessageListenerAdapter foreclosureListenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //监听价格趋势和市场情况主题并绑定消息订阅处理器
        container.addMessageListener(priceListenerAdapter, new PatternTopic("PRICE_TOPIC"));
        //监听挂牌案例主题并绑定消息订阅处理器
        container.addMessageListener(listingListenerAdapter, new PatternTopic("LISTING_TOPIC"));
        //监听交易案例主题并绑定消息订阅处理器
        container.addMessageListener(caseListenerAdapter, new PatternTopic("CASE_TOPIC"));
        //监听法拍案例主题并绑定消息订阅处理器
        container.addMessageListener(foreclosureListenerAdapter, new PatternTopic("FORECLOSURE_TOPIC"));
        return container;
    }

    /**
     * 价格趋势和市场消息订阅处理器,并指定处理方法
     * @param receiver
     * @return
     */
    @Bean
    MessageListenerAdapter priceListenerAdapter(ReceiverRedisMessage receiver) {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "priceReceive");
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = this.jacksonSerializer();
        messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        return messageListenerAdapter;
    }

    /**
     * 挂牌案例消息订阅处理器,并指定处理方法
     * @param receiver
     * @return
     */
    @Bean
    MessageListenerAdapter listingListenerAdapter(ReceiverRedisMessage receiver) {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "listingReceive");
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = this.jacksonSerializer();
        messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        return messageListenerAdapter;
    }


    /**
     * 交易案例消息订阅处理器,并指定处理方法
     * @param receiver
     * @return
     */
    @Bean
    MessageListenerAdapter caseListenerAdapter(ReceiverRedisMessage receiver) {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "caseReceive");
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = this.jacksonSerializer();
        messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        return messageListenerAdapter;
    }


    /**
     * 法拍案例消息订阅处理器,并指定处理方法
     * @param receiver
     * @return
     */
    @Bean
    MessageListenerAdapter foreclosureListenerAdapter(ReceiverRedisMessage receiver) {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "foreclosureReceive");
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = this.jacksonSerializer();
        messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        return messageListenerAdapter;
    }

    @Bean
    ReceiverRedisMessage receiver(CountDownLatch latch) {
        return new ReceiverRedisMessage(latch);
    }

    @Bean
    CountDownLatch latch() {
        return new CountDownLatch(1);
    }

    @Bean
    RedisTemplate template(RedisTemplate redisTemplate) {
        return redisTemplate;
    }

    private Jackson2JsonRedisSerializer jacksonSerializer(){
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        return jackson2JsonRedisSerializer;
    }
}

 

import com.alibaba.fastjson.JSON;
import com.cindata.esp.domian.casehistory.CaseHistory;
import com.cindata.esp.domian.foreclosure.ForeclosureHistory;
import com.cindata.esp.domian.netlisting.NetListingHistory;
import com.cindata.esp.domian.pricetendency.PriceTendencySituation;
import com.cindata.esp.service.ICaseService;
import com.cindata.esp.service.IForeclosureService;
import com.cindata.esp.service.INetListingService;
import com.cindata.esp.service.IPirceTendencyService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.concurrent.CountDownLatch;

@Slf4j
public class ReceiverRedisMessage {

    private CountDownLatch latch;

    @Autowired
    IPirceTendencyService priceTendencyService;

    @Autowired
    INetListingService netListingService;

    @Autowired
    ICaseService caseService;

    @Autowired
    IForeclosureService foreclosureService;

    @Autowired
    public ReceiverRedisMessage(CountDownLatch latch) {
        this.latch = latch;
    }

    /**
     * 价格趋势和市场队列消息接收方法
     * @param jsonMsg
     */
    public void priceReceive( String jsonMsg) {
    log.info("[开始消费REDIS消息队列PRICE_TOPIC数据...]");
    try{
        PriceTendencySituation priceTendencySituation = JSON.parseObject(jsonMsg, PriceTendencySituation.class);
        priceTendencyService.save(priceTendencySituation);
        log.info("[消费REDIS消息队列PRICE_TOPIC数据成功.]");
    }catch(Exception e){
        log.error("[消费REDIS消息队列PRICE_TOPIC数据失败，失败信息:{}]", e.getMessage());
    }
    latch.countDown();
    }


    /**
     * 挂牌案例队列消息接收方法
     * @param jsonMsg
     */
    public void listingReceive( String jsonMsg) {
    log.info("[开始消费REDIS消息队列LISTING_TOPIC数据...]");
    try{
        NetListingHistory netListingHistory = JSON.parseObject(jsonMsg, NetListingHistory.class);
        netListingService.save(netListingHistory);
        log.info("[消费REDIS消息队列LISTING_TOPIC数据成功.]");
    }catch(Exception e){
        log.error("[消费REDIS消息队列LISTING_TOPIC数据失败，失败信息:{}]", e.getMessage());
    }
    latch.countDown();
    }

    /**
     * 交易案例队列消息接收方法
     * @param jsonMsg
     */
    public void caseReceive( String jsonMsg) {
    log.info("[开始消费REDIS消息队列CASE_TOPIC数据...]");
    try{
        CaseHistory caseHistory = JSON.parseObject(jsonMsg, CaseHistory.class);
        caseService.save(caseHistory);
        log.info("[消费REDIS消息队列CASE_TOPIC数据成功.]");
    }catch(Exception e){
        log.error("[消费REDIS消息队列CASE_TOPIC数据失败，失败信息:{}]", e.getMessage());
    }
    latch.countDown();
    }

    /**
     * 法拍案例队列消息接收方法
     * @param jsonMsg
     */
    public void foreclosureReceive( String jsonMsg) {
    log.info("[开始消费REDIS消息队列FORECLOSURE_TOPIC数据...]");
    try{
        ForeclosureHistory foreclosureHistory = JSON.parseObject(jsonMsg, ForeclosureHistory.class);
        foreclosureService.save(foreclosureHistory);
        log.info("[消费REDIS消息队列FORECLOSURE_TOPIC数据成功.]");
    }catch(Exception e){
        log.error("[消费REDIS消息队列FORECLOSURE_TOPIC数据失败，失败信息:{}]", e.getMessage());
    }
    latch.countDown();
    }
}

这样，应用启动时，消息的订阅方(subscriber)就注册好了。这时候只要使用一个简单的程序，模拟publisher，向指定topic发布消息，RedisMessageListener就可以接收到消息，spring-redis的写法是这样：

redisTemplate.convertAndSend("PRICE_TOPIC", "hello world!");

发布订阅模式，在集群化部署时会重复消费数据，暂时还没找到一个类似消费kafka数据手工维护偏移量的方法。