基于Redis实现消息队列的示例代码_Redis

消息队列在分布式系统中非常重要，能够有效解耦系统的各个模块，提供异步处理能力和缓冲能力。redis作为一个高性能的内存数据库，除了缓存和持久化存储，它还能充当轻量级的消息队列。使用redis处理消息队列有助于提高系统的吞吐量和可扩展性。

一、使用场景

消息队列的应用场景非常广泛，包括：

异步任务处理：如发送邮件、短信、推送通知等耗时操作，可以通过消息队列异步执行，提升用户体验。
系统解耦：将生产者与消费者解耦，使得两个系统无需直接通信，互相独立。
流量削峰：在高并发场景下，通过消息队列对请求进行排队处理，缓解系统的压力峰值。
日志处理：可以将日志消息推送到队列中，集中处理和存储。

二、原理解析

redis提供了几种不同的机制来实现消息队列，包括list和pub/sub。

1. 基于list的消息队列

redis的list数据结构是实现队列的基础。常见的操作包括：

lpush：将消息推入队列的左端。
rpush：将消息推入队列的右端。
rpop：从队列的右端弹出消息（相当于先进先出，即fifo）。
blpop：阻塞式弹出消息，当队列为空时会等待直到有新的消息。

2. 基于pub/sub的发布订阅

redis的**发布/订阅（pub/sub）**是一种不同的消息队列实现方式，支持消息广播。它的机制如下：

发布者发布消息到一个频道（channel）。
所有订阅了该频道的消费者都能接收到消息。

但pub/sub的特点是消息不持久化，它更适用于实时消息传递，如果没有订阅者，消息会丢失。

三、实现过程

1. 项目结构

我们的项目基于spring boot ，包括以下模块：

producer：消息生产者，用于将任务或消息推入队列。
consumer：消息消费者，负责从队列中读取任务并处理。

2. 环境准备

在pom.xml中添加redis和web的依赖：

<dependencies>
    <dependency>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-data-redis</artifactid>
    </dependency>
    <dependency>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-web</artifactid>
    </dependency>
</dependencies>

在application.yml中配置redis：

spring:
  redis:
    host: localhost
    port: 6379

3. redis配置类

配置redistemplate用于与redis进行交互：

@configuration
public class redisconfig {
    @bean
    public redistemplate<string, object> redistemplate(redisconnectionfactory redisconnectionfactory) {
        redistemplate<string, object> template = new redistemplate<>();
        template.setconnectionfactory(redisconnectionfactory);
        return template;
    }
}

4. 基于list的消息队列实现

producer（消息生产者）

生产者将消息推入队列中，使用lpush或rpush操作：

@service
public class messageproducer {

    @autowired
    private redistemplate<string, object> redistemplate;

    private static final string message_queue = "message:queue";

    public void produce(string message) {
        redistemplate.opsforlist().leftpush(message_queue, message);
    }
}

consumer（消息消费者）

消费者从队列中阻塞式地弹出消息，并进行处理：

@service
public class messageconsumer {

    @autowired
    private redistemplate<string, object> redistemplate;

    private static final string message_queue = "message:queue";

    @scheduled(fixedrate = 5000) // 每5秒检查一次队列
    public void consume() {
        string message = (string) redistemplate.opsforlist().rightpop(message_queue);
        if (message != null) {
            system.out.println("consumed message: " + message);
            // 模拟处理消息
        }
    }
}

通过@scheduled注解，消费者可以定期从redis队列中拉取消息进行处理。

5. 基于pub/sub的消息队列实现

producer（发布者）

发布者将消息发布到指定频道：

@service
public class pubsubproducer {

    @autowired
    private redistemplate<string, object> redistemplate;

    public void publishmessage(string channel, string message) {
        redistemplate.convertandsend(channel, message);
    }
}

consumer（订阅者）

订阅者监听频道的消息并处理：

@service
public class pubsubconsumer implements messagelistener {

    @override
    public void onmessage(message message, byte[] pattern) {
        system.out.println("received message: " + new string(message.getbody()));
    }
}

redis配置订阅监听器

配置订阅器并注册频道：

@configuration
public class redispubsubconfig {

    @bean
    public messagelisteneradapter messagelistener() {
        return new messagelisteneradapter(new pubsubconsumer());
    }

    @bean
    public redismessagelistenercontainer rediscontainer(redisconnectionfactory connectionfactory,
                                                        messagelisteneradapter listeneradapter) {
        redismessagelistenercontainer container = new redismessagelistenercontainer();
        container.setconnectionfactory(connectionfactory);
        container.addmessagelistener(listeneradapter, new patterntopic("pubsub:channel"));
        return container;
    }
}

6. controller层

为生产者提供api接口：

@restcontroller
@requestmapping("/queue")
public class queuecontroller {

    @autowired
    private messageproducer messageproducer;

    @autowired
    private pubsubproducer pubsubproducer;

    // 将消息放入队列
    @postmapping("/produce")
    public responseentity<string> producemessage(@requestparam string message) {
        messageproducer.produce(message);
        return responseentity.ok("message produced");
    }

    // 发布消息
    @postmapping("/publish")
    public responseentity<string> publishmessage(@requestparam string message) {
        pubsubproducer.publishmessage("pubsub:channel", message);
        return responseentity.ok("message published");
    }
}

四、测试效果

基于list的消息队列：
- 启动spring boot应用后，通过api接口发送消息：
  - post请求：/queue/produce
  - 参数：message=helloqueue
- 消费者将在每次调度时从队列中取出消息并打印。
基于pub/sub的消息队列：
- 发布消息：
  - post请求：/queue/publish
  - 参数：message=hellopubsub
- 订阅者将立即收到消息并处理。