【redis】stream消息队列

news2024/11/16 22:42:30

目录

        • 截图
        • 一、代码示例
          • 1.1 pom.xml依赖
          • 1.2 application.xml配置
          • 1.3 启动类
          • 1.4 配置类
          • 1.5 消息实体
          • 1.6 自定义错误
          • 1.7 自定义注解
          • 1.8 服务类
          • 1.9 监听类
          • 1.10 controller

截图

在这里插入图片描述
在这里插入图片描述

一、代码示例

1.1 pom.xml依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>springboot-learning</artifactId>
        <groupId>com.learning</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>springboot-mq-redis</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-redis</artifactId>
            <version>2.5.11</version>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-data-26</artifactId>
            <version>3.17.1</version>
        </dependency>
        <!--fastjson-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.1.15</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>
1.2 application.xml配置
spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password:
    database: 0
    ssl: false
redis:
  model: 1
1.3 启动类
package com.learning.mq.redis;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @Description 启动类
 */

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
1.4 配置类
package com.learning.mq.redis.config;

import lombok.Data;

@Data
public class RedisClusterProperties {

    /**
     * 集群状态扫描间隔时间，单位是毫秒
     */
    private int scanInterval=5000;

    /**
     * 集群节点
     */
    private String nodes;

    /**
     * 默认值： SLAVE（只在从服务节点里读取）设置读取操作选择节点的模式。 可用值为： SLAVE - 只在从服务节点里读取。
     * MASTER - 只在主服务节点里读取。 MASTER_SLAVE - 在主从服务节点里都可以读取
     */
    private String readMode;
    /**
     * （从节点连接池大小） 默认值：64
     */
    private int slaveConnectionPoolSize;
    /**
     * 主节点连接池大小）默认值：64
     */
    private int masterConnectionPoolSize;

    /**
     * （命令失败重试次数） 默认值：3
     */
    private int retryAttempts;

    /**
     *命令重试发送时间间隔，单位：毫秒 默认值：1500
     */
    private int retryInterval;

    /**
     * 执行失败最大次数默认值：3
     */
    private int failedAttempts;

}

package com.learning.mq.redis.config;

import lombok.Data;

@Data
public class RedisPoolProperties {

    private int maxIdle;

    private int minIdle;

    private int maxActive;

    private int maxWait;

    private int connTimeout;

    private int soTimeout = 30000;
    /**
     * 池大小
     */
    private  int size;

}


package com.learning.mq.redis.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Component
@ConfigurationProperties(prefix = "spring.redis")
@Data
public class RedisProperties {

    private int database;

    /**
     * 等待节点回复命令的时间。该时间从命令发送成功时开始计时
     */
    private int timeout;

    private String password;

    /**
     * 池配置
     */
    private RedisPoolProperties pool;

    /**
     * 集群 信息配置
     */
    private RedisClusterProperties cluster;

    /**
     * 哨兵配置
     */
    private RedisSentinelProperties sentinel;

    private  String host;
    private  String port;

    public String getAddress() {
        return this.getHost()+":"+this.getPort();
    }

}


package com.learning.mq.redis.config;

import lombok.Data;

@Data
public class RedisSentinelProperties {

    /**
     * 哨兵master 名称
     */
    private String master;

    /**
     * 哨兵节点
     */
    private String nodes;

    /**
     * 哨兵配置
     */
    private boolean masterOnlyWrite;

    /**
     *
     */
    private int failMax;

}


package com.learning.mq.redis.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.*;
import org.redisson.spring.data.connection.RedissonConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@Configuration
@EnableConfigurationProperties(RedisProperties.class)
@Slf4j
public class RedissonConfig {
    @Autowired
    private RedisProperties redisProperties;
    @Value("${redis.model}")
    private String mode;
    @Value("${spring.redis.database:0}")
    private String database;

    /**
     * 单机模式 redisson 客户端
     */
    @Bean
    @ConditionalOnProperty(name = "redis.model", havingValue = "1")
    public RedissonClient redissonSingle() {
        log.info("redis.model=1, 为单机模式");
        Config config = new Config();
        String node = redisProperties.getAddress();
        node = node.startsWith("redis://") ? node : "redis://" + node;
        SingleServerConfig serverConfig = config.useSingleServer()
                .setAddress(node)
                .setDatabase(redisProperties.getDatabase());
//                .setTimeout(redisProperties.getPool().getConnTimeout());
//                .setConnectionPoolSize(redisProperties.getPool().getSize())
//                .setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());
        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }
        return Redisson.create(config);
    }

    /**
     * 哨兵模式 redisson 客户端
     *
     * @return
     */

    @Primary
    @Bean
    @ConditionalOnProperty(name = "redis.model", havingValue = "2")
    public RedissonClient redissonSentinel() {
        log.info("redis.model=2, 为哨兵模式");
        Config config = new Config();
        String[] nodes = redisProperties.getSentinel().getNodes().split(",");
        List<String> newNodes = new ArrayList(nodes.length);
        Arrays.stream(nodes).forEach((index) -> newNodes.add(index.startsWith("redis://") ? index : "redis://" + index));
        SentinelServersConfig serverConfig = config.useSentinelServers()
                .addSentinelAddress(newNodes.toArray(new String[0]))
                .setDatabase(redisProperties.getDatabase())
                .setMasterName(redisProperties.getSentinel().getMaster())
                .setReadMode(ReadMode.MASTER)//TODO 如果实现读写分离时，需要配置这里
                .setTimeout(redisProperties.getTimeout());
//                .setMasterConnectionPoolSize(redisProperties.getPool().getSize())
//                .setSlaveConnectionPoolSize(redisProperties.getPool().getSize());
        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }
        return Redisson.create(config);
    }

    @Primary
    @Bean("redissonConnectionFactory")
    public RedissonConnectionFactory redissonConnectionFactory() {
        RedissonClient redisson;
        if ("1".equals(mode)) {
            redisson = redissonSingle();
        } else if ("2".equals(mode)) {
            redisson = redissonSentinel();
        } else {
            mode = "0";
            redisson = redissonCluster();
        }
        return new RedissonConnectionFactory(redisson);
    }

    /**
     * 集群模式的 redisson 客户端
     *
     * @return
     */
    @Bean
    @ConditionalOnProperty(name = "redis.model", havingValue = "3")
    RedissonClient redissonCluster() {
        log.info("redis.model=3, 为集群模式");
        Config config = new Config();
        String[] nodes = redisProperties.getCluster().getNodes().split(",");
        List<String> newNodes = new ArrayList(nodes.length);
        Arrays.stream(nodes).forEach((index) -> newNodes.add(index.startsWith("redis://") ? index : "redis://" + index));

        ClusterServersConfig serverConfig = config.useClusterServers()
                .addNodeAddress(newNodes.toArray(new String[0]))
                .setScanInterval(redisProperties.getCluster().getScanInterval())
                .setReadMode(ReadMode.MASTER)
//                .setIdleConnectionTimeout(redisProperties.getPool().getSoTimeout())
//                .setConnectTimeout( redisProperties.getPool().getConnTimeout())
//                .setRetryAttempts( redisProperties.getCluster().getRetryAttempts())
//                .setRetryInterval( redisProperties.getCluster().getRetryInterval())
//                .setMasterConnectionPoolSize(redisProperties.getCluster().getMasterConnectionPoolSize())
//                .setSlaveConnectionPoolSize(redisProperties.getCluster().getSlaveConnectionPoolSize())
                .setPingConnectionInterval(1000)
                .setTimeout(redisProperties.getTimeout());
        if (StringUtils.isNotBlank(redisProperties.getPassword())) {
            serverConfig.setPassword(redisProperties.getPassword());
        }
        return Redisson.create(config);
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        // 使用Jackson2JsonRedisSerialize 替换默认序列化
        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);

        // 设置value的序列化规则和 key的序列化规则
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);

        redisTemplate.setDefaultSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setEnableDefaultSerializer(true);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        StringRedisTemplate template = new StringRedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }
}
1.5 消息实体
package com.learning.mq.redis.dto;

import lombok.Data;

/**
 * @Description 消息实体
 */
@Data
public class MessageDTO {
    private String content;
}
1.6 自定义错误
package com.learning.mq.redis.handler;

import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ErrorHandler;

/**
 * @description
 */
@Slf4j
public class CustomerErrorHandler implements ErrorHandler {
    @Override
    public void handleError(Throwable e) {
        //TODO 消费异常时处理逻辑
        log.error("发生了异常", e);
    }
}
1.7 自定义注解
package com.learning.mq.redis.annotation;

import java.lang.annotation.*;

/**
 * 消息队列自定义注解
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface RedisStreamMessageQueueListener {
    /**
     * 消息队列的 key
     * @return
     */
    String value();

    /**
     * 消费者组
     * @return
     */
    String group() default "";

    /**
     * 消息反序列化类型
     * @return
     */
    Class type();

    /**
     * 是否自动 acknowledge
     * @return
     */
    boolean autoAck() default true;

    /**
     * 每次拉取的消息数,线程池每秒拉取一次
     * @return
     */
    int batchSize() default 5;

    /**
     * 启用死信队列
     * @return
     */
    boolean enableDeadLetter() default false;

    /**
     * 投递次数阈值
     * 超过阈值后，消息将被添加到死信队列中
     * @return
     */
    int maxDeliverTimes() default 3;

    /**
     * 消息未Ack时间阈值（单位：秒）
     * 超过阈值的消息将被重新投递到其他消费者（仅当消费者组存在多个消费者情况）
     * @return
     */
    int timeout() default 300;

    /**
     * 死信队列的key，当enableDeadLetter=true时必填
     * @return
     */
    String deadLetterKey() default "";

    /**
     * 死信队列最大长度
     * @return
     */
    int deadLetterMaxSize() default 1000;

}
1.8 服务类
package com.learning.mq.redis.service;

import com.alibaba.fastjson.JSON;
import com.learning.mq.redis.annotation.RedisStreamMessageQueueListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.dao.DataAccessException;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStreamCommands;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ListenerAnnotation implements ApplicationListener<ApplicationStartedEvent>, InitializingBean {
    protected Logger streamMqLog = LoggerFactory.getLogger(ListenerAnnotation.class);

    private static final String DISTUIBUTE_HANDLE_PENDING_DATA_LOCK = "distuibute_handle_pending_data_lock";

    private static List<RedisStreamMessageQueueListener> ans = new ArrayList<>();
    // 记录违规的死信队列keys（违规：死信队列又配置了死信队列）
    private List<String> illegalDeadLetterKeys = new ArrayList<>();

    private volatile static ListenerAnnotation listenerAnnotation;

    private final RedisStreamMessageQueueStartService redisStreamMessageQueueStartService;
    private RedisTemplate redisTemplate;
    private final String applicationName;
    private final RedissonClient redissonClient;

    // 当前服务ip及端口号，用于动态消费者组的游标继承
    private final String serviceIP;
    private final String servicePort;

    private ListenerAnnotation(RedisStreamMessageQueueStartService redisStreamMessageQueueStartService, RedisTemplate redisTemplate, String applicationName, RedissonClient redissonClient, String serviceIP, String servicePort) {
        this.redisStreamMessageQueueStartService = redisStreamMessageQueueStartService;
        this.redisTemplate = redisTemplate;
        this.applicationName = applicationName;
        this.redissonClient = redissonClient;
        this.serviceIP = serviceIP;
        this.servicePort = servicePort;
    }

    /**
     * 单例
     *
     * @param redisStreamMessageQueueStartService
     * @return
     */
    public static ListenerAnnotation getListenerAnnotation(RedisStreamMessageQueueStartService redisStreamMessageQueueStartService, RedisTemplate redisTemplate, String group, RedissonClient redissonClient, String serviceIP, String servicePort) {
        if (listenerAnnotation == null) {
            synchronized (ListenerAnnotation.class) {
                if (listenerAnnotation == null) {
                    listenerAnnotation = new ListenerAnnotation(redisStreamMessageQueueStartService, redisTemplate, group, redissonClient, serviceIP, servicePort);
                }
            }
        }
        return listenerAnnotation;
    }

    /**
     * 循环处理keys中的pending数据
     */
    private void handlePendingData() {
        for (RedisStreamMessageQueueListener redisStreamMessageQueueListener : ans) {
            String key = redisStreamMessageQueueListener.value();
            String groupName = redisStreamMessageQueueListener.group();
            StreamInfo.XInfoGroups groups = redisTemplate.opsForStream().groups(key);
            if (groups.groupCount() > 0) {
                Iterator<StreamInfo.XInfoGroup> groupIterator = groups.stream().iterator();
                while (groupIterator.hasNext()) {
                    StreamInfo.XInfoGroup group = groupIterator.next();
                    if (StringUtils.equals(group.groupName(), groupName)) {
                        StreamInfo.XInfoConsumers consumers = redisTemplate.opsForStream().consumers(key, groupName);
                        if (group.pendingCount() > 0) {
                            PendingMessagesSummary pendingMessagesSummary = redisTemplate.opsForStream().pending(key, groupName);
                            PendingMessages pendingMessages = redisTemplate.opsForStream().pending(key, group.groupName(), pendingMessagesSummary.getIdRange(), 100);
                            if (pendingMessages.size() > 0) {
                                for (PendingMessage pendingMessage : pendingMessages) {
                                    if (pendingMessage.getElapsedTimeSinceLastDelivery().toMillis() > redisStreamMessageQueueListener.timeout() * 1000) {
                                        final String recordId = pendingMessage.getId().getValue();
                                        streamMqLog.info("redis stream listen task. mqConfig: [{}] - stream: [{}] - group: [{}] - recordId: [{}] 发现消息超过限定时长[{}]s", redisStreamMessageQueueListener, key, group.groupName(), recordId, redisStreamMessageQueueListener.timeout());
                                        try {
                                            Map recordVal = null;
                                            // 查询消息内容
                                            List<MapRecord<String, Object, Object>> readRes = redisTemplate.opsForStream().range(key, Range.closed(recordId, recordId), new RedisZSetCommands.Limit().count(1));
                                            if (readRes != null && readRes.size() > 0) {
                                                for (MapRecord<String, Object, Object> v : readRes) {
                                                    if (StringUtils.equals(v.getId().getValue(), recordId)) {
                                                        recordVal = v.getValue();
                                                    }
                                                }
                                            }
                                            if (recordVal != null) {
                                                streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 查询到消息的内容为[{}]", key, group.groupName(), recordId, JSON.toJSONString(recordVal));
                                                if (pendingMessage.getTotalDeliveryCount() <= redisStreamMessageQueueListener.maxDeliverTimes() && consumers.size() > 1) {
                                                    streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - consumers: [{}]", key, group.groupName(), consumers);
                                                    // 移交其他消费者处理
                                                    String newConsumer = consumers.get(consumers.size() - 1).consumerName();
                                                    for (Iterator<StreamInfo.XInfoConsumer> consumerIterator = consumers.stream().iterator(); ; consumerIterator.hasNext()) {
                                                        StreamInfo.XInfoConsumer curConsumer = consumerIterator.next();
                                                        if (StringUtils.equals(curConsumer.consumerName(), pendingMessage.getConsumerName())) {
                                                            break;
                                                        } else {
                                                            newConsumer = curConsumer.consumerName();
                                                        }
                                                    }
                                                    if (StringUtils.isNotBlank(newConsumer)) {
                                                        streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 从消费者[{}]移交给消费者[{}]处理", key, group.groupName(), recordId, pendingMessage.getConsumerName(), newConsumer);
                                                        final String newConsumerFinal = newConsumer;
                                                        redisTemplate.execute(new RedisCallback<List<ByteRecord>>() {
                                                            @Override
                                                            public List<ByteRecord> doInRedis(RedisConnection connection) throws DataAccessException {
                                                                RedisStreamCommands.XClaimOptions claimOptions = RedisStreamCommands.XClaimOptions.minIdle(Duration.ofSeconds(1)).ids(pendingMessage.getId());
                                                                return connection.streamCommands().xClaim(key.getBytes(), groupName, newConsumerFinal, claimOptions.idle(Duration.ofSeconds(1)));
                                                            }
                                                        });
                                                    } else {
                                                        streamMqLog.warn("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 移交消费者失败，未查询到有效的新消费者！", key, group.groupName(), recordId);
                                                    }
                                                } else {
                                                    // 记录日志或者放入配置好的队列中
                                                    if (redisStreamMessageQueueListener.enableDeadLetter()) {
                                                        String deadLetterKey = redisStreamMessageQueueListener.deadLetterKey();
                                                        if (StringUtils.isNotBlank(deadLetterKey) && !illegalDeadLetterKeys.contains(deadLetterKey) && redisStreamMessageQueueListener.deadLetterMaxSize() > 0) {
                                                            // 放入队列中
                                                            redisStreamMessageQueueStartService.coverSend(deadLetterKey, recordVal, new Integer(redisStreamMessageQueueListener.deadLetterMaxSize()).longValue());
                                                            streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 放入死信队列[{}]中", key, group.groupName(), recordId, deadLetterKey);
                                                        } else {
                                                            // 报错违规
                                                            streamMqLog.warn("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 放入死信队列[{}]失败！原因：死信队列为空或者违规套用死信队列，或未配置死信队列最大长度", key, group.groupName(), recordId, deadLetterKey);
                                                            if (streamMqLog.isDebugEnabled()) {
                                                                streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] - data: [{}]", key, group.groupName(), recordId, JSON.toJSONString(recordVal));
                                                            }
                                                        }
                                                    }
                                                }
                                            } else {
                                                streamMqLog.warn("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 未查询到消息的内容", key, group.groupName(), recordId);
                                            }
                                        } catch (Exception e) {
                                            log.error("", e);
                                            streamMqLog.error("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 处理数据失败，原因：{}", key, group.groupName(), recordId, e.getMessage());
                                        } finally {
                                            // ack数据
                                            Long ackLen = redisTemplate.opsForStream().acknowledge(key, groupName, recordId);
                                            streamMqLog.info("redis stream listen task. stream: [{}] - group: [{}] - recordId: [{}] 消息ack[{}]条", key, group.groupName(), recordId, ackLen);
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    /**
     * 添加监听
     *
     * @param event
     */
    @Override
    public void onApplicationEvent(ApplicationStartedEvent event) {
        Map<String, Object> beans = event.getApplicationContext().getBeansWithAnnotation(RedisStreamMessageQueueListener.class);
        List<String> deadLetterKeys = new ArrayList<>();
        for (Object bean : beans.values()) {
            RedisStreamMessageQueueListener ca = bean.getClass().getAnnotation(RedisStreamMessageQueueListener.class);
            String group = ca.group();
            redisStreamMessageQueueStartService.listener(ca.value(), group, ca.type(), ca.autoAck(), ca.batchSize(), (StreamListener) bean);
            if (StringUtils.isNotBlank(group)) {
                ans.add(ca);
            }
            if (ca.enableDeadLetter() && StringUtils.isNotBlank(ca.deadLetterKey())) {
                deadLetterKeys.add(ca.deadLetterKey());
            }
        }
        // 获取违规死信队列（违规：死信队列又配置了死信队列）
        ans.forEach(v -> {
            if (v.enableDeadLetter() && StringUtils.isNotBlank(v.deadLetterKey()) && deadLetterKeys.contains(v.value())) {
                illegalDeadLetterKeys.add(v.value());
            }
        });
    }

    @Override
    public void afterPropertiesSet() {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            RLock lock = redissonClient.getLock(DISTUIBUTE_HANDLE_PENDING_DATA_LOCK + applicationName);
            boolean res = false;
            try {
                res = lock.tryLock(1, 10, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.error("", e);
            }
            if (!res) {
                return;
            }
            try {
                this.handlePendingData();
            } catch (Exception e) {
                log.error("", e);
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                } else {
                    log.warn("Lock not held by CurrentThread!!!");
                }
            }
        },
        0, 10, TimeUnit.SECONDS);
    }
}


package com.learning.mq.redis.service;

import org.springframework.data.redis.stream.StreamListener;

/**
 * 监听和发送方法
 */
public interface RedisStreamMessageQueueStartService {
    /**
     * 添加监听器
     * @param event
     * @param groupName
     * @param type
     * @param autoAck
     * @param batchSize
     * @param streamListener
     */
    void listener(String event, String groupName, Class type, boolean autoAck, int batchSize, StreamListener streamListener);

    <V> void coverSend(String event, V val, Long maxSize);
}


package com.learning.mq.redis.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import com.learning.mq.redis.handler.CustomerErrorHandler;

import java.net.Inet4Address;
import java.net.UnknownHostException;
import java.time.Duration;

@Slf4j
public class RedisStreamMessageQueueStartServiceImpl implements RedisStreamMessageQueueStartService {

    private final long dataCenterId = getDataCenterId();

    private final RedisTemplate<String, Object> redisTemplate;

    long maxLen;

    public RedisStreamMessageQueueStartServiceImpl(RedisTemplate<String, Object> redisTemplate, Long maxLen) {
        this.redisTemplate = redisTemplate;
        this.maxLen = maxLen;
    }

    @Override
    public void listener(String event, String groupName, Class type, boolean autoAck, int batchSize, StreamListener streamListener) {
        // 如果是动态消费者组，则无需创建消费者组
        if (StringUtils.isNotBlank(groupName)) {
            createGroup(event, groupName);
        }
        startSubscription(event, groupName, type, autoAck, batchSize, streamListener);
    }

    @Override
    public <V> void coverSend(String event, V val, Long maxSize) {
        ObjectRecord<String, V> record = StreamRecords.newRecord()
                .ofObject(val)
                .withId(RecordId.autoGenerate())
                .withStreamKey(event);
        redisTemplate.opsForStream().add(record);
        redisTemplate.opsForStream().trim(event, maxSize, true);
        if (log.isDebugEnabled()) {
            log.debug("event {} send content {}", event, val);
        }
    }


    private void startSubscription(String event, String groupName, Class type, boolean autoAck, int batchSize, StreamListener streamListener) {
        RedisConnectionFactory redisConnectionFactory = redisTemplate.getConnectionFactory();
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .targetType(type)
                .batchSize(batchSize)
                .hashValueSerializer(RedisSerializer.string())
                .objectMapper(new ObjectHashMapper())
                .errorHandler(new CustomerErrorHandler())
                .build();
        StreamMessageListenerContainer listenerContainer = StreamMessageListenerContainer
                .create(redisConnectionFactory, options);
        if (StringUtils.isBlank(groupName)) {
            listenerContainer.receive(
                    StreamOffset.create(event, ReadOffset.lastConsumed()),
                    streamListener);
        } else {
            if (autoAck) {
                listenerContainer.receiveAutoAck(
                        Consumer.from(groupName, groupName + dataCenterId),
                        StreamOffset.create(event, ReadOffset.lastConsumed()),
                        streamListener);
            } else {
                listenerContainer.receive(
                        Consumer.from(groupName, groupName + dataCenterId),
                        StreamOffset.create(event, ReadOffset.lastConsumed()),
                        streamListener);
            }
        }
        listenerContainer.start();
    }

    private void createGroup(String event, String groupName) {
        try {
            String addGroupResult = redisTemplate.opsForStream().createGroup(event, ReadOffset.latest(), groupName);
            if (!StringUtils.equals(addGroupResult, "OK")) {
                log.error("STREAM - Group create failed");
            }
        } catch (Exception e) {
            log.info("STREAM - Group create failed. reason: exist group.");
        }
    }

    private static Long getDataCenterId() {
        try {
            String hostName = Inet4Address.getLocalHost().getHostName();
            int[] ints = StringUtils.toCodePoints(hostName);
            int sums = 0;
            for (int b : ints) {
                sums += b;
            }
            return (long) (sums % 32);
        } catch (UnknownHostException e) {
            // 如果获取失败，则使用随机数备用
            return RandomUtils.nextLong(0, 31);
        }
    }
}
1.9 监听类
package com.learning.mq.redis.listener;

import com.alibaba.fastjson.JSON;
import com.learning.mq.redis.annotation.RedisStreamMessageQueueListener;
import com.learning.mq.redis.dto.MessageDTO;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

@Component
@RedisStreamMessageQueueListener(value = "message-stream", group = "consumerGroup", type = String.class, timeout = 20, autoAck = true)
@AllArgsConstructor
@Slf4j
public class MessageListener implements StreamListener<String, ObjectRecord<String, String>> {

    private final RedisTemplate redisTemplate;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        try {
            String stream = message.getStream();
            RecordId id = message.getId();
            String msg = message.getValue();
//            MessageDTO messageDTO = JSON.parseObject(msg, MessageDTO.class);
            log.info("消费的消息:{}", msg);
            //当是消费组消费时，如果不是自动ack，则需要在这个地方手动ack
//          redisTemplate.opsForStream().acknowledge("consumerGroup", message);
            //移除消息
            redisTemplate.opsForStream().delete(stream, id.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
1.10 controller
package com.learning.mq.redis.controller;

import lombok.AllArgsConstructor;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Description 消息发送
 */
@RestController
@RequestMapping("/message")
@AllArgsConstructor
public class MessageController {
    private RedisTemplate redisTemplate;

    @GetMapping("/save")
    public String save(@RequestParam String message){
//        try {
            String key = "message-stream";
//            MessageDTO messageDTO = new MessageDTO();
//            messageDTO.setContent(message);
//            ObjectRecord<String, MessageDTO> record = StreamRecords.newRecord()
//                    .in(key)
//                    .ofObject(messageDTO)
//                    .withId(RecordId.autoGenerate());
//            String msgId = redisTemplate.opsForStream().add(record).toString();
//            return msgId;
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//        return message;
        ObjectRecord<String, String> record = StreamRecords.newRecord()
                .in(key)
                .ofObject(message)
                .withId(RecordId.autoGenerate());
        String msgId = redisTemplate.opsForStream().add(record).toString();
        return msgId;
    }
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/700627.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Unity VR：XR Interaction Toolkit 官方 Demo

Unity VR：XR Interaction Toolkit 官方 Demo

Unity XR Interaction Toolkit 提供了一个官方 Demo&#xff0c;包含了丰富的功能演示&#xff0c;可以供大家更好地学习 VR 开发。 项目地址&#xff1a;https://github.com/Unity-Technologies/XR-Interaction-Toolkit-Examples 项目里包括多个演示场景&#xff0c;而 XRI_…
阅读更多...
Unity 中 TextMeshPro 字体位置偏上/偏下

Unity 中 TextMeshPro 字体位置偏上/偏下

问题&#xff1a;在Unity中创建了一个新的TextMeshPro 字体&#xff0c;在使用的时候布局设置的居中对齐&#xff0c;但在场景中实际位置却和预期位置不服&#xff0c;如下图。 当然通过调整布局设置&#xff0c;也可以显示成正常的效果&#xff0c;但不可能所有文本都通过这…
阅读更多...
element栅格自定义等分

element栅格自定义等分

由于element栅格组件是 24 分栏&#xff0c;所以当我们需要5、7、9等分时&#xff0c;就不好实现 <template><el-row><el-col :span"24">1</el-col></el-row><el-row><el-col :span"12">2</el-col><e…
阅读更多...
【Java可执行命令】（三）API文档生成工具javadoc： 深入解析Java API文档生成工具javadoc ~

【Java可执行命令】（三）API文档生成工具javadoc： 深入解析Java API文档生成工具javadoc ~

Java可执行命令详解之javadoc 1️⃣ 概念2️⃣ 优势和缺点3️⃣ 使用3.1 语法格式3.1.1 可选参数&#xff1a;-d < directory>3.1.2 可选参数&#xff1a;-sourcepath < pathlist>3.1.3 可选参数&#xff1a;-classpath < pathlist>3.1.4 可选参数&#xff1…
阅读更多...
JTAG 、 SWD 和 J-Link、ST-Link

JTAG 、 SWD 和 J-Link、ST-Link

JTAG和SWD的区别与联系JTAG接口SWD接口JTAG和SWD的区别与联系J-Link和ST-LinkJ-LINK仿真器STLINK仿真器JLINK和STLINK的比较与选择 JTAG和SWD的区别与联系 JTAG和SWD是两种常用的用于调试和编程ARM微控制器的接口&#xff0c;它们都可以通过调试器&#xff08;如ST-LINK或J-Li…
阅读更多...
插入排序——希尔排序

插入排序——希尔排序

希尔排序其实就是一种插入排序&#xff0c;实际上就是通过直接插入排序一步步改进优化而实现的。所以在了解希尔排序之前要明白插入排序的实现原理。 插入排序 其实我觉得插入排序也可以叫做摸牌排序&#xff0c;就是从第二张牌开始处理&#xff0c;将摸到的牌按照合适的顺序插…
阅读更多...
音频分析仪-测试

音频分析仪-测试

底噪&#xff1a; 有效值&#xff1a; ** 增益&#xff1a; ** ** 延时-脉冲响应 **
阅读更多...
基于Java+Swing+Mysql影院售票系统

基于Java+Swing+Mysql影院售票系统

基于JavaSwingMysql影院售票系统 一、系统介绍二、功能展示1.主页2.新增影片信息3.删除影片信息 三、数据库四、其他系统实现五、获取源码 一、系统介绍 该系统实现了查看影片列表、新增影片信息、删除影片信息 运行环境&#xff1a;eclipse、idea、jdk1.8 二、功能展示 1.…
阅读更多...
大环境之下软件测试行业趋势能否上升？

大环境之下软件测试行业趋势能否上升？

如果说&#xff0c;2022年对于全世界来说&#xff0c;都是一场极大的挑战的话&#xff1b;那么&#xff0c;2023年绝对是机遇多多的一年。众所周知&#xff0c;随着疫情在全球范围内逐步得到控制&#xff0c;无论是国际还是国内的环境&#xff0c;都会呈现逐步回升的趋势&#…
阅读更多...
Java教程-Java异常传播

Java教程-Java异常传播

异常首先从调用堆栈的顶部抛出&#xff0c;如果没有被捕获&#xff0c;它会向下传递到前一个方法。如果在那里没有被捕获&#xff0c;异常会再次向下传递到前一个方法&#xff0c;依此类推&#xff0c;直到它们被捕获或者达到调用堆栈的最底部。这被称为异常传播。 异常传播示例…
阅读更多...
Beego之Bee安装以及创建,运行项目

Beego之Bee安装以及创建,运行项目

一.简介 Bee是什么&#xff1f; bee工具是一个为了协助快速开发 Beego 项目而创建的项目&#xff0c;通过 bee 可以很容易的进行 Beego 项目的 创建、热编译、开发、测试和部署 Beego中文文档 Beego中文文档: Beego简介 安装前提 在安装bee之前&#xff0c;首先得提前安装好Go的…
阅读更多...
5.6.1 端口及套接字

5.6.1 端口及套接字

5.6.1 端口及套接字 传输层的作用是在通信子网提供服务的基础之上为它的上层也就是应用进程提供端到端的传输服务&#xff0c;通信子网是由用作信息交换的网络节点和通信线路所组成的独立的数据通信系统。它承担着全网的数据传输、转接和加工变换等通信处理工作。如图 通信子网…
阅读更多...
【前端工程化】Verdaccio搭建本地npm仓库

【前端工程化】Verdaccio搭建本地npm仓库

背景 Verdaccio 是一个 Node.js创建的轻量的私有npm proxy registry 我们在开发npm包的时候&#xff0c;经常需要验证发包流程&#xff0c;或者开发的npm包仅局限于公司内部使用时&#xff0c;就可以借助Verdaccio搭建一个npm仓库&#xff0c;搭建完之后&#xff0c;只要更改np…
阅读更多...
力扣 700. 二叉搜索树中的搜索

力扣 700. 二叉搜索树中的搜索

题目来源&#xff1a;https://leetcode.cn/problems/search-in-a-binary-search-tree/description/ C题解1&#xff1a;二叉搜索树&#xff0c;右节点大于当前节点&#xff0c;左右节点小于当前节点&#xff0c;因此可以根据当前节点值与目标值的大小比较进行搜索。 class Sol…
阅读更多...
【CSS】鼠标(移入/移出)平滑(显示/隐藏)下划线

【CSS】鼠标(移入/移出)平滑(显示/隐藏)下划线

文章目录 效果展示实现步骤1. 添加背景颜色2. 修改背景颜色3. 调整背景的大小4. 取消背景重复绘制5. 调小高度6. 设置背景绘制位置7. 隐藏背景8. 加入鼠标移入事件9. 平滑显示/隐藏下划线10. 调整一下背景图的位置11. 调整鼠标移入时进入的位置 效果展示 鼠标移入内容时&#…
阅读更多...
基于matlab使用二维规范化互相关进行模式匹配和目标跟踪（附源码）

基于matlab使用二维规范化互相关进行模式匹配和目标跟踪（附源码）

一、前言 此示例演示如何使用二维规范化互相关进行模式匹配和目标跟踪。该示例使用预定义或用户指定的目标以及要跟踪的类似目标的数量。归一化互相关图显示&#xff0c;当值超过设置的阈值时&#xff0c;将标识目标。 在此示例中&#xff0c;您使用规范化互相关来跟踪视频中…
阅读更多...
行业云“组合拳”+AIGC开放战略，新华三的精耕务实之道

行业云“组合拳”+AIGC开放战略，新华三的精耕务实之道

“今年或许不是实现宏伟目标的一年&#xff0c;但却是重新聚焦、重新调整和重新思考基础设施的时刻。”这是Gartner研究副总裁Paul Delory在谈到影响2023年云、数据中心和边缘基础设施趋势时所表达的观点&#xff0c;而影响趋势之一就是云团队将优化和重构云基础设施。对于企业…
阅读更多...
爬虫入门指南：Python网络请求及常见反爬虫策略应对方法

爬虫入门指南：Python网络请求及常见反爬虫策略应对方法

文章目录 引言HTTP协议与请求方法HTTP协议请求方法 使用Python进行网络请求安装Requests库发送GET请求发送POST请求 反爬虫与应对策略IP限制使用代理IP&#xff1a; 用户代理检测设置User-Agent头部&#xff1a; 验证码参考方案 动态页面请求频率限制未完待续.... 引言 在当今…
阅读更多...
1.盒子模型

1.盒子模型

页面布局要学习三大核心&#xff0c;盒子模型&#xff0c;浮动和定位.学习好盒子模型能非常好的帮助我们布局页面. 1.1看透网页布局的本质 网页布局过程: 1.先准备好相关的网页元素,网页元素基本都是盒子 2.利用CSS设置好盒子样式,然后摆放到相应位置 3.往盒子里面装内容. 网…
阅读更多...
自定义MVC框架【上篇】--原理

自定义MVC框架【上篇】--原理

&#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 接下来看看由辉辉所写的关于自定义MVC的相关操作吧 目录 &#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 一.什么是自定义MVC框架&#xff1f; 二…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023