当前位置：首页 > article >正文

java实现redis的消息发送和消费，类似kafka功能

article 2024/10/20 18:33:03

确保在 pom.xml 中添加了 Spring Data Redis 和 Jedis 的依赖。如下所示：

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

配置 Redis 连接

使用 RedisTemplate 进行操作

在 Spring Boot 中，推荐使用 RedisTemplate 进行 Redis 操作，而不是直接使用 Jedis。下面是如何创建和使用 RedisTemplate 的示例。

创建 RedisConfig 类
创建一个配置类来定义 RedisTemplate 的 Bean

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;

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        return template;
    }
}

使用 RedisTemplate 发送和接收消息

接下来，可以使用 RedisTemplate 进行消息的发送和接收。

发送消息
创建一个消息生产者的类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void sendMessage(String queueName, String message) {
        redisTemplate.opsForList().leftPush(queueName, message);
        System.out.println("Message sent to queue: " + message);
    }
}

接收消息创建一个消息消费者的类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class MessageConsumer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void consumeMessage(String queueName) {
        String message = (String) redisTemplate.opsForList().rightPop(queueName);
        if (message != null) {
            System.out.println("Message received from queue: " + message);
        } else {
            System.out.println("No message in queue.");
        }
    }
}

使用 Publish/Subscribe 模式

对于使用发布/订阅模式，可以创建一个订阅类，并使用 Spring 的 @MessageMapping 注解或直接使用
JedisPubSub。

发布者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class Publisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void publishMessage(String channelName, String message) {
        redisTemplate.convertAndSend(channelName, message);
        System.out.println("Message published to channel " + channelName + ": " + message);
    }
}

订阅者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

@Service
public class Subscriber {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void subscribe(String channelName) {
        Jedis jedis = (Jedis) redisTemplate.getConnectionFactory().getConnection().getNativeConnection();
        jedis.subscribe(new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("Message received from channel " + channel + ": " + message);
            }
        }, channelName);
    }
}