脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot实现Redis Stream队列

SpringBoot中实现Redis Stream队列的代码实例

 更新时间：2024年09月12日 10:14:19   作者：保加利亚的风  
本文介绍了如何在Spring Boot中使用Redis Stream队列进行消息的生产和消费,涉及到的主要内容包括添加Redis依赖、配置RedisTemplate、创建生产者和消费者监听器等,需要的朋友可以参考下

前言

简单实现一下在SpringBoot中操作Redis Stream队列的方式，监听队列中的消息进行消费。

  • jdk：1.8
  • springboot-version：2.6.3
  • redis：5.0.1（5版本以上才有Stream队列）

准备工作

1、pom

redis 依赖包（version 2.6.3）

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

2、 yml

spring: 
  redis:
    database: 0
    host: 127.0.0.1

3、 RedisStreamUtil工具类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
public class RedisStreamUtil {

	@Autowired
	private RedisTemplate<String, Object> redisTemplate;

	/**
	 * 创建消费组
	 *
	 * @param key   键名称
	 * @param group 组名称
	 * @return {@link String}
	 */
	public String oup(String key, String group) {
		return redisTemplate.opsForStream().createGroup(key, group);
	}

	/**
	 * 获取消费者信息
	 *
	 * @param key   键名称
	 * @param group 组名称
	 * @return {@link StreamInfo.XInfoConsumers}
	 */
	public StreamInfo.XInfoConsumers queryConsumers(String key, String group) {
		return redisTemplate.opsForStream().consumers(key, group);
	}

	/**
	 * 查询组信息
	 *
	 * @param key 键名称
	 * @return
	 */
	public StreamInfo.XInfoGroups queryGroups(String key) {
		return redisTemplate.opsForStream().groups(key);
	}

	// 添加Map消息
	public String addMap(String key, Map<String, Object> value) {
		return redisTemplate.opsForStream().add(key, value).getValue();
	}

	// 读取消息
	public List<MapRecord<String, Object, Object>> read(String key) {
		return redisTemplate.opsForStream().read(StreamOffset.fromStart(key));
	}

	// 确认消费
	public Long ack(String key, String group, String... recordIds) {
		return redisTemplate.opsForStream().acknowledge(key, group, recordIds);
	}

	// 删除消息。当一个节点的所有消息都被删除，那么该节点会自动销毁
	public Long del(String key, String... recordIds) {
		return redisTemplate.opsForStream().delete(key, recordIds);
	}

	// 判断是否存在key
	public boolean hasKey(String key) {
		Boolean aBoolean = redisTemplate.hasKey(key);
		return aBoolean != null && aBoolean;
	}
}

代码实现

生产者发送消息

生产者发送消息，在Service层创建addMessage方法，往队列中发送消息。

代码中addMap()方法第一个参数为key，第二个参数为value，该key要和后续配置的保持一致，暂时先记住这个key。

@Service
@Slf4j
@RequiredArgsConstructor
public class RedisStreamMqServiceImpl implements RedisStreamMqService {

    private final RedisStreamUtil redisStreamUtil;

    /**
     * 发送一个消息
     *
     * @return {@code Object}
     */
    @Override
    public Object addMessage() {
        RedisUser redisUser = new RedisUser();
        redisUser.setAge(18);
        redisUser.setName("hcr");
        redisUser.setEmail("156ef561@gmail.com");

        Map<String, Object> message = new HashMap<>();
        message.put("user", redisUser);

        String recordId = redisStreamUtil.addMap("mystream", message);
        return recordId;
    }
}

controller接口方法

@RestController
@RequestMapping("/redis")
@Slf4j
@RequiredArgsConstructor
public class RedisController {

    private final RedisStreamMqService redisStreamMqService;

    @GetMapping("/addMessage")
    public Object addMessage() {
        return redisStreamMqService.addMessage();
    }
}

调用测试，查看redis中是否正常添加数据。

接口返回数据

1702622585248-0

查看redis中的数据

消费者监听消息进行消费

创建RedisConsumersListener监听器

import cn.hcr.utils.RedisStreamUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
@RequiredArgsConstructor
public class RedisConsumersListener implements StreamListener<String, MapRecord<String, String, String>> {

    public final RedisStreamUtil redisStreamUtil;

    /**
     * 监听器
     *
     * @param message
     */
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // stream的key值
        String streamKey = message.getStream();
        //消息ID
        RecordId recordId = message.getId();
        //消息内容
        Map<String, String> msg = message.getValue();
        log.info("【streamKey】= " + streamKey + ",【recordId】= " + recordId + ",【msg】=" + msg);

        //处理逻辑

        //逻辑处理完成后，ack消息，删除消息，group为消费组名称
        StreamInfo.XInfoGroups xInfoGroups = redisStreamUtil.queryGroups(streamKey);
        xInfoGroups.forEach(xInfoGroup -> redisStreamUtil.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));
        redisStreamUtil.del(streamKey, recordId.getValue());
    }
}

创建RedisConfig配置类，配置监听

package cn.hcr.config;

import cn.hcr.listener.RedisConsumersListener;
import cn.hcr.utils.RedisStreamUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
@Slf4j
public class RedisConfig {

    @Resource
    private RedisStreamUtil redisStreamUtil;

    /**
     * redis序列化
     *
     * @param redisConnectionFactory
     * @return {@code RedisTemplate<String, Object>}
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }

    @Bean
    public Subscription subscription(RedisConnectionFactory factory) {
        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName("async-stream-consumer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer
                        .StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(5)
                        .executor(executor)
                        .pollTimeout(Duration.ofSeconds(1))
                        .errorHandler(throwable -> {
                            log.error("[MQ handler exception]", throwable);
                            throwable.printStackTrace();
                        })
                        .build();
        
        //该key和group可根据需求自定义配置
        String streamName = "mystream";
        String groupname = "mygroup";

        initStream(streamName, groupname);
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        // 手动ask消息
        Subscription subscription = listenerContainer.receive(Consumer.from(groupname, "zhuyazhou"),
                StreamOffset.create(streamName, ReadOffset.lastConsumed()), new RedisConsumersListener(redisStreamUtil));
        // 自动ask消息
           /* Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from(redisMqGroup.getName(), redisMqGroup.getConsumers()[0]),
                    StreamOffset.create(streamName, ReadOffset.lastConsumed()), new ReportReadMqListener());*/
        listenerContainer.start();
        return subscription;
    }

    private void initStream(String key, String group) {
        boolean hasKey = redisStreamUtil.hasKey(key);
        if (!hasKey) {
            Map<String, Object> map = new HashMap<>(1);
            map.put("field", "value");
            //创建主题
            String result = redisStreamUtil.addMap(key, map);
            //创建消费组
            redisStreamUtil.oup(key, group);
            //将初始化的值删除掉
            redisStreamUtil.del(key, result);
            log.info("stream:{}-group:{} initialize success", key, group);
        }
    }
}

redisTemplate：该bean用于配置redis序列化

subscription：配置监听

initStream：初始化消费组

监听测试

使用addMessage()方法投送一条消息后，查看控制台输出信息。

【streamKey】= mystream,
【recordId】= 1702623008044-0,
【msg】=
{user=[
    "cn.hcr.pojo.RedisUser",
    {"name":"hcr","age":18,"email":"156ef561@gmail.com"}
    ]
}

总结

以上就是在SpringBoot中简单实现Redis Stream队列的Demo，如有需要源码或者哪里不清楚的请评论或者发送私信。
Template：该bean用于配置redis序列化

subscription：配置监听

initStream：初始化消费组

到此这篇关于SpringBoot中实现Redis Stream队列的文章就介绍到这了,更多相关SpringBoot实现Redis Stream队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Java并发 synchronized锁住的内容解析

    Java并发 synchronized锁住的内容解析

    这篇文章主要介绍了Java并发 synchronized锁住的内容解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-09-09
  • springboot在filter中如何用threadlocal存放用户身份信息

    springboot在filter中如何用threadlocal存放用户身份信息

    这篇文章主要介绍了springboot中在filter中如何用threadlocal存放用户身份信息,本文章主要描述通过springboot的filter类,在过滤器中设置jwt信息进行身份信息保存的方法,需要的朋友可以参考下
    2024-07-07
  • Java中的原生post请求方式

    Java中的原生post请求方式

    这篇文章主要介绍了Java中的原生post请求方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-10-10
  • SpringBoot+React实现计算个人所得税

    SpringBoot+React实现计算个人所得税

    本文将以个人所得税的计算为例,使用React+SpringBoot+GcExcel来实现这一功能,文中的示例代码讲解详细,具有一定的学习价值,感兴趣的小伙伴可以了解下
    2023-09-09
  • Spring实战之SpEl语法实例详解

    Spring实战之SpEl语法实例详解

    这篇文章主要介绍了Spring实战之SpEl语法,结合实例形式分析了SpEl创建数组、集合及解析变量等相关操作原理与实现技巧,需要的朋友可以参考下
    2019-12-12
  • SpringBoot 自动配置原理及源码解析

    SpringBoot 自动配置原理及源码解析

    SpringBoot 在项目启动的时候封装了创建对象的方法，无需我们手动配置,接下来通过本文给大家介绍SpringBoot 自动配置原理解析及源码展示，感兴趣的朋友一起看看吧
    2021-06-06
  • SpringBoot项目设置断点debug调试无效忽略web.xml问题的解决

    SpringBoot项目设置断点debug调试无效忽略web.xml问题的解决

    这篇文章主要介绍了SpringBoot项目设置断点debug调试无效忽略web.xml问题的解决，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2019-08-08
  • Java编程利用socket多线程访问服务器文件代码示例

    Java编程利用socket多线程访问服务器文件代码示例

    这篇文章主要介绍了Java编程利用socket多线程访问服务器文件代码示例，具有一定参考价值，需要的朋友可以了解下。
    2017-10-10
  • Java的@Transactional、@Aysnc、事务同步问题详解

    Java的@Transactional、@Aysnc、事务同步问题详解

    这篇文章主要介绍了Java的@Transactional、@Aysnc、事务同步问题详解,现在我们需要在一个业务方法中插入一个用户,这个业务方法我们需要加上事务,然后插入用户后,我们要异步的方式打印出数据库中所有存在的用户,需要的朋友可以参考下
    2023-11-11
  • Java中使用byte[]获取16位字符串的技巧分享

    Java中使用byte[]获取16位字符串的技巧分享

    在Java编程中,处理字符编码和字节数组时,尤其是从​​byte[]​​​数组中获取字符串,可能会遇到字符集的复杂性问题,本文将介绍如何从一个​​byte[]​​数组中正确获取16位字符串,需要的朋友可以参考下
    2024-08-08

最新评论