spring boot + Redis实现消息队列-生产消费者

实现思路：

Redis本身提供了一个发布/订阅模式，但生产消费者模式需要我们自己去实现。

利用Redis中的队列，将新消息放入名称为xx的队列末尾，完成消息生产者。
启动一个线程，使用 brpop 命令循环从xx队列取第一个元素，获得消息，调用注册的消费者执行业务逻辑。

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

配置RedisTemplate

/**
 * redis配置
 *
 */
@Configuration
public class RedisConfig {


    /**
     * retemplate相关配置
     * @param factory
     * @return
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        // 配置连接工厂
        template.setConnectionFactory(factory);

        //使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值（默认使用JDK的序列化方式）
        Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);

        ObjectMapper om = new ObjectMapper();
        // 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jacksonSeial.setObjectMapper(om);

        // 值采用json序列化
        template.setValueSerializer(jacksonSeial);
        //使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());

        // 设置hash key 和value序列化模式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(jacksonSeial);
        template.afterPropertiesSet();

        return template;
    }

    /**
     * 对hash类型的数据操作
     *
     * @param redisTemplate
     * @return
     */
    @Bean
    public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForHash();
    }

    /**
     * 对redis字符串类型数据操作
     *
     * @param redisTemplate
     * @return
     */
    @Bean
    public ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForValue();
    }

    /**
     * 对链表类型的数据操作
     *
     * @param redisTemplate
     * @return
     */
    @Bean
    public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForList();
    }

    /**
     * 对无序集合类型的数据操作
     *
     * @param redisTemplate
     * @return
     */
    @Bean
    public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForSet();
    }

    /**
     * 对有序集合类型的数据操作
     *
     * @param redisTemplate
     * @return
     */
    @Bean
    public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForZSet();
    }

    /**
     * redis消息监听器容器
     * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
     * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
     * @param connectionFactory
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter webMessage) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //订阅了一个叫chat的通道
        container.addMessageListener(webMessage, new PatternTopic("webMessage"));

        return container;
    }

    @Bean
    MessageListenerAdapter webMessage(MessageReceiver messageReceiver) {

        //给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”
        //不填defaultListenerMethod默认调用handleMessage
        return new MessageListenerAdapter(messageReceiver, "receiveMessage");
    }

    /**
     * 配置redis消息队列消费者容器
     *
     * @param redisTemplate redis
     * @return 消费者容器
     */
    @Autowired
    NormalMessageReceiver normalMessageReceiver;


    @Bean(initMethod = "init", destroyMethod = "destroy")
    public RedisMqConsumerContainer redisMqConsumerContainer(RedisTemplate<String, Object> redisTemplate) {
        RedisMqConsumerContainer config = new RedisMqConsumerContainer(redisTemplate);
        config.addConsumer(QueueConfiguration.builder()
                .queue("normalMessage")
                .consumer(normalMessageReceiver)
                .build());

        return config;
    }

消息-生产者

public class QueueSender {
    private RedisTemplate<Object, Object> redisTemplate;

    public QueueSender(RedisTemplate<Object, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void sendMsg(String queue, Object msg) {
        redisTemplate.opsForList().leftPush(queue, msg);

    }
}

消息-消费者

消费接口类

public interface MsgConsumer {
    void onMessage(Object message);

    void onError(Object msg, Exception e);
}

消息监听类

class QueueListener implements Runnable {
    public static final Logger log = LoggerFactory.getLogger(QueueListener.class);
    private RedisTemplate<Object, Object> redisTemplate;
    private String queue;
    private MsgConsumer consumer;

    public QueueListener(RedisTemplate<Object, Object> redisTemplate, String queue, MsgConsumer consumer) {
        this.redisTemplate = redisTemplate;
        this.queue = queue;
        this.consumer = consumer;
    }

    @Override
    public void run() {
        log.info("QueueListener start...queue:{}", queue);
        while (RedisMqConsumerContainer.run) {
            try {
                Object msg = redisTemplate.opsForList().rightPop(queue, 30, TimeUnit.SECONDS);
                if (msg != null) {
                    try {
                        consumer.onMessage(msg);
                    } catch (Exception e) {
                        consumer.onError(msg, e);
                    }
                }
            } catch (QueryTimeoutException ignored) {
            } catch (Exception e) {
                if (RedisMqConsumerContainer.run) {
                    log.error("Queue:{}", queue, e);
                } else {
                    log.info("QueueListener exits...queue:{}", queue);
                }
            }
        }
    }
}

消息-消费者容器

配置类

public class QueueConfiguration {
    /**
     * 队列名称
     */
    private String queue;
    /**
     * 消费者
     */
    private MsgConsumer consumer;

    private QueueConfiguration() {
    }

    public static Builder builder() {
        return new Builder();
    }

    String getQueue() {
        return queue;
    }

    MsgConsumer getConsumer() {
        return consumer;
    }

    public static class Builder {
        private QueueConfiguration configuration = new QueueConfiguration();

        public QueueConfiguration defaultConfiguration(MsgConsumer consumer) {
            configuration.consumer = consumer;
            configuration.queue = consumer.getClass().getSimpleName();
            return configuration;
        }

        public Builder queue(String queue) {
            configuration.queue = queue;
            return this;
        }

        public Builder consumer(MsgConsumer consumer) {
            configuration.consumer = consumer;
            return this;
        }

        public QueueConfiguration build() {
            if (configuration.queue == null || configuration.queue.length() == 0) {
                if (configuration.consumer != null) {
                    configuration.queue = configuration.getClass().getSimpleName();
                }
            }
            return configuration;
        }

    }
}

消息--消费者容器

public class RedisMqConsumerContainer {
    private static final Logger log = LoggerFactory.getLogger(RedisMqConsumerContainer.class);
    private Map<String, QueueConfiguration> consumerMap = new HashMap<>();
    private RedisTemplate<Object, Object> redisTemplate;
    static boolean run;
    private ExecutorService exec;

    public RedisMqConsumerContainer(RedisTemplate<Object, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void addConsumer(QueueConfiguration configuration) {
        if (consumerMap.containsKey(configuration.getQueue())) {
            log.warn("Key:{} this key already exists, and it will be replaced", configuration.getQueue());
        }
        if (configuration.getConsumer() == null) {
            log.warn("Key:{} consumer cannot be null, this configuration will be skipped", configuration.getQueue());
        }
        consumerMap.put(configuration.getQueue(), configuration);
    }

    public void destroy() {
        run = false;
        this.exec.shutdown();
        log.info("QueueListener exiting.");
        while (!this.exec.isTerminated()) {

        }
        log.info("QueueListener exited.");
    }

    public void init() {
        run = true;
        this.exec = Executors.newCachedThreadPool(r -> {
            final AtomicInteger threadNumber = new AtomicInteger(1);
            return new Thread(r, "RedisMQListener-" + threadNumber.getAndIncrement());
        });
        consumerMap = Collections.unmodifiableMap(consumerMap);
        consumerMap.forEach((k, v) -> exec.submit(new QueueListener(redisTemplate, v.getQueue(), v.getConsumer())));
    }

}

配置消费者

public class NormalMessageReceiver implements MsgConsumer {
    private static Logger log = LoggerFactory.getLogger(TestListener.class);

    @Override
    public void onMessage(Object message) {
        log.info("收到消息:" + message);
        
    }

    @Override
    public void onError(Object msg, Exception e) {
        log.error("发生错误,消息:{}", msg, e);
    }
}

测试代码如下：

//开启定时器功能
@EnableScheduling
@Component
public class MessageSender {

    @Autowired
    private QueueSender queueSender;

    @Scheduled(fixedDelay = 5000)
    public void sendMessageXXX() {
        //stringRedisTemplate.convertAndSend("chat1", String.valueOf(Math.random()));
        //producer.publishMessageXXX("队列消息1xxxxx"+System.currentTimeMillis());

        for (int i = 0; i < 20; i++) {
            queueSender.sendMsg("TEST0","hello quit~~~~,序号:"+i);
            queueSender.sendMsg("TEST1","hello quit~~~~,序号:"+i);
            queueSender.sendMsg("TEST2","hello quit~~~~,序号:"+i);
        }

    }
}

spring boot + Redis实现消息队列-生产消费者

实现思路：

配置RedisTemplate

消息-生产者

消息-消费者

消费接口类

消息监听类

消息-消费者容器

配置类

消息--消费者容器

配置消费者

测试代码如下：

继续阅读

Protobuf 和 Thrift对比（转）

如何解决Redis缓存击穿、雪崩、穿透问题

signal 6 (SIGABRT), code -6 name: RenderThread问题记录

java 版本的redis-stat不能运行在后台和daemon

功能强大的Gson部分功能用法

redis管理常用命令

django短信验证码的后端实现

Redis订阅了一段时间后订阅失效了（ redisTemplate.convertAndSend）

2022秋招面试总结（cpp+java+测开）百度测开一面字节后端一面虾皮后端一面虾皮后端二面

数据迁移方法数据迁移原则数据迁移之双写方案数据迁移之级联同步方案

微服务-性能压测\缓存redis和分布式锁redisson和SpringCache

Nacos 2.0 升级前后性能对比压测

Spring数据和Redis

redis集群数据一致性_RedisRaft为Redis集群带来强大的数据一致性

supervisor 管理redis 和httpd 环境centos7

Redis简介一(单机版)发展历程Redis