redis5 stream消息机制java封装实现

TkMessage 消息体

public class TkMessage extends AbstractEntry {
    
    private GroupTopicEnum groupTopic;
    
    private String tag;
    
    private String key;
    
    private String content;

    public GroupTopicEnum getGroupTopic() {
        return groupTopic;
    }

    public void setGroupTopic(GroupTopicEnum groupTopic) {
        this.groupTopic = groupTopic;
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    public String getKey() {
        return key;
    }

    public void setKey(String key) {
        this.key = key;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

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

    public static class Builder {
        
        private GroupTopicEnum groupTopic;
        
        private String tag;
        
        private String key;
        
        private String content;

        public TkMessage build() {
            TkMessage message = new TkMessage();
            message.setGroupTopic(this.groupTopic);
            message.setTag(this.tag);
            message.setKey(this.key);
            message.setContent(this.content);
            return message;
        }

        public Builder setGroupTopic(GroupTopicEnum groupTopic) {
            this.groupTopic = groupTopic;
            return this;
        }

        public Builder setTag(String tag) {
            this.tag = tag;
            return this;
        }

        public Builder setKey(String key) {
            this.key = key;
            return this;
        }

        public Builder setContent(String content) {
            this.content = content;
            return this;
        }

        public GroupTopicEnum getGroupTopic() {
            return groupTopic;
        }

        public String getTag() {
            return tag;
        }

        public String getKey() {
            return key;
        }

        public String getContent() {
            return content;
        }
    }
}

AbstractEntry

public class AbstractEntry implements Serializable {
    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
    }
}

GroupTopicEnum

public class GroupTopicEnum extends AbstractEntry {
    
    private String group;
    
    private String topic;

    public GroupTopicEnum() {
    }

    public GroupTopicEnum(String group, String topic) {
        this.group = group;
        this.topic = topic;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public boolean equals(GroupTopicEnum obj) {
        boolean r = super.equals(obj);
        if (!r) {
            r = this.group.equals(obj.getGroup()) && this.topic.equals(obj.getTopic());
        }
        return r;
    }
}

MessageProducer

package io.shulie.jmeter.tool.redis.message;

import io.shulie.jmeter.tool.redis.RedisConfig;
import io.shulie.jmeter.tool.redis.domain.TkMessage;
import io.shulie.jmeter.tool.redis.util.JsonUtil;
import io.shulie.jmeter.tool.redis.util.MessageUtil;
import io.shulie.jmeter.tool.redis.util.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.*;



public class MessageProducer {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private static final Map producerMap = new HashMap<>();
    private StringRedisTemplate stringRedisTemplate;

    public static MessageProducer getInstance(RedisConfig redisConfig) {
        if (null == redisConfig) {
            return null;
        }
        MessageProducer producer = producerMap.get(redisConfig);
        if (null == producer) {
            producer = new MessageProducer();
            producer.init(redisConfig);
            producerMap.put(redisConfig, producer);
        }
        return producer;
    }

    public static MessageProducer getInstance(RedisConnectionFactory factory) {
        if (null == factory) {
            return null;
        }
        MessageProducer producer = producerMap.get(factory);
        if (null == producer) {
            producer = new MessageProducer();
            producer.init(factory);
            producerMap.put(factory, producer);
        }
        return producer;
    }

    public static MessageProducer getInstance(StringRedisTemplate stringRedisTemplate) {
        if (null == stringRedisTemplate) {
            return null;
        }
        MessageProducer producer = producerMap.get(stringRedisTemplate);
        if (null == producer) {
            producer = new MessageProducer();
            producer.setStringRedisTemplate(stringRedisTemplate);
            producerMap.put(stringRedisTemplate, producer);
        }
        return producer;
    }

    public boolean send(TkMessage message) throws Exception {
        if (null == message) {
            return false;
        }
        if (null == message.getGroupTopic()) {
            throw new Exception("message groupTopicTag is null!");
        }
        String streamKey = MessageUtil.buildRedisMessageStreamKey(message.getGroupTopic());

        // 创建消息记录, 以及指定stream
        StringRecord stringRecord = StreamRecords.string(Collections.singletonMap("value", JsonUtil.toJson(message))).withStreamKey(streamKey);
        // 将消息添加至消息队列中
        RecordId recordId = null;
        try {
            recordId = this.stringRedisTemplate.opsForStream().add(stringRecord);
        } catch (Exception e) {
            logger.error("send message failed!message="+ JsonUtil.toJson(message), e);
        }
        return null != recordId;
    }

    private void init(RedisConfig redisConfig) {
        if (null == redisConfig) {
            return;
        }
        init(RedisUtil.getJedisConnectionFactory(redisConfig));
    }

    private void init(RedisConnectionFactory factory) {
        if (null == factory) {
            return;
        }
        stringRedisTemplate = RedisUtil.getStringRedisTemplate(factory);
    }

    public StringRedisTemplate getStringRedisTemplate() {
        return stringRedisTemplate;
    }

    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

}

IMessage

public interface IMessageConsumerListener {
    
    public GroupTopicEnum getGroupTopic();
    
    public String group();
    
    public String getConsumer();

    
    public List getTags();
    
    public boolean receive(TkMessage message);
}

AbstractMessageConsumerListener

public abstract class AbstractMessageConsumerListener implements IMessageConsumerListener {

    @Override
    public String group() {
        return "default";
    }

    @Override
    public String getConsumer() {
        return "consumer-1";
    }
}

MessageConsumer

package io.shulie.jmeter.tool.redis.message;

import io.shulie.jmeter.tool.redis.RedisConfig;
import io.shulie.jmeter.tool.redis.domain.TkMessage;
import io.shulie.jmeter.tool.redis.util.JsonUtil;
import io.shulie.jmeter.tool.redis.util.MessageUtil;
import io.shulie.jmeter.tool.redis.util.RedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;


public class MessageConsumer implements StreamListener> {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
    public static long DEATH_LINE = 300;
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private static final Map consumerMap = new HashMap<>();

    private Executor executor;
    private final Map containerMap = new HashMap<>();
    private RedisConfig config;
    private StringRedisTemplate stringRedisTemplate;

    public static MessageConsumer getInstance(RedisConfig config) {
        if (null == config) {
            return null;
        }
        MessageConsumer consumer = consumerMap.get(config);
        if (null == consumer) {
            consumer = new MessageConsumer();
            consumer.init(config);
            consumerMap.put(config, consumer);
        }
        return consumer;
    }
    public static MessageConsumer getInstance(RedisConnectionFactory factory) {
        if (null == factory) {
            return null;
        }
        MessageConsumer consumer = consumerMap.get(factory);
        if (null == consumer) {
            consumer = new MessageConsumer();
            consumer.init(factory);
            consumerMap.put(factory, consumer);
        }
        return consumer;
    }

    public static MessageConsumer getInstance(StringRedisTemplate stringRedisTemplate) {
        if (null == stringRedisTemplate) {
            return null;
        }
        MessageConsumer consumer = consumerMap.get(stringRedisTemplate);
        if (null == consumer) {
            consumer = new MessageConsumer();
            consumer.setStringRedisTemplate(stringRedisTemplate);
            consumerMap.put(stringRedisTemplate, consumer);
        }
        return consumer;
    }

    public MessageConsumer setThreadPool(Executor executor) {
        this.executor = executor;
        return this;
    }

    public MessageConsumer register(IMessageConsumerListener listener) {
        if (null == executor) {
            throw new RuntimeException("请先初始化线程池");
        }
        if (null == listener.getGroupTopic()) {
            throw new RuntimeException("未定义groupTopic");
        }
        StreamMessageListenerContainer> container = containerMap.get(listener);
        if (null == container) {
            container = buildStreamContainer(listener);
            container.start();
            containerMap.put(listener, container);
        }
        return this;
    }

    @Override
    public void onMessage(ObjectRecord message) {
        try{
            // 消息的key和value
            TkMessage data = JsonUtil.parseObject(message.getValue(), TkMessage.class);
            if (null == data) {
                logger.error("message反序列化结果为空：message.value="+message.getValue());
                return;
            }

            containerMap.keySet().stream().filter(Objects::nonNull)
                .filter(listener -> listener.getGroupTopic().equals(data.getGroupTopic()))
                .filter(listener -> MessageUtil.matchTags(listener, data))
                .map(listener -> (Runnable) () -> {
                    if (listener.receive(data)) {
                        try {
                            String group = MessageUtil.getValue("default", listener, IMessageConsumerListener::group);
                            this.stringRedisTemplate.opsForStream().acknowledge(group, message);
                        } catch (Exception e) {
                            logger.error("message ack errror!message="+JsonUtil.toJson(data), e);
                        }
                    }
                })
                .forEach(executor::execute);
        } catch (Exception e) {
            // 处理异常
            logger.error("onMessage error!", e);
        }
    }

    public void destory() {
        containerMap.values().stream().filter(Objects::nonNull).forEach(container -> {
            try {
                container.stop();
            } catch (Exception e) {
                logger.warn("container stop failed!", e);
            }
        });
    }

    private void init(RedisConfig config) {
        if (null == config) {
            return;
        }
        this.config = config;
        init(RedisUtil.getJedisConnectionFactory(config));
    }

    private void init(RedisConnectionFactory factory) {
        if (null == factory) {
            return;
        }
        stringRedisTemplate = RedisUtil.getStringRedisTemplate(factory);
        //定时任务，死信消息定时清理
        scheduledExecutorService.scheduleWithFixedDelay(this::scanPendingMsg, 5L, 5L, TimeUnit.SECONDS);
    }

    private StreamMessageListenerContainer> buildStreamContainer(IMessageConsumerListener listener) {
        String streamKey = MessageUtil.buildRedisMessageStreamKey(listener.getGroupTopic());
        String group = MessageUtil.getValue("default", listener, IMessageConsumerListener::group);
        RedisUtil.checkAndCreateStreamGroup(stringRedisTemplate, streamKey, group);
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .pollTimeout(Duration.ofSeconds(5)) // 拉取消息超时时间
                        .batchSize(10) // 批量抓取消息
                        .targetType(String.class) // 传递的数据类型
                        .build();
        StreamMessageListenerContainer> container = StreamMessageListenerContainer.create(stringRedisTemplate.getConnectionFactory(), options);
        //指定消费最新的消息
        StreamOffset offset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());
        //创建消费者1
        Consumer consumer1 = Consumer.from(group, StringUtils.isBlank(listener.getConsumer()) ? "consumer1" : listener.getConsumer());
        StreamMessageListenerContainer.StreamReadRequest streamReadRequest = StreamMessageListenerContainer.StreamReadRequest.builder(offset)
                .errorHandler((error) -> {
                    logger.error("build streamReadRequest error!", error);
                })
                .cancelOnError(e -> false)
                .consumer(consumer1)
                //关闭自动ack确认
                .autoAcknowledge(false)
                .build();
        //指定消费者对象
        container.register(streamReadRequest, this);

        return container;
    }

    public Map getContainerMap() {
        return containerMap;
    }

    public RedisConfig getConfig() {
        return config;
    }

    public StringRedisTemplate getStringRedisTemplate() {
        return stringRedisTemplate;
    }

    public static Map getConsumerMap() {
        return consumerMap;
    }

    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    
    public void scanPendingMsg() {
        containerMap.keySet().stream().filter(Objects::nonNull)
            .forEach(this::scanPendingMsg);
    }

    private void scanPendingMsg(IMessageConsumerListener listener) {
        if (null == listener) {
            return;
        }
        String streamKey = MessageUtil.buildRedisMessageStreamKey(listener.getGroupTopic());

        StreamOperations streamOperations = this.stringRedisTemplate.opsForStream();
        String group = MessageUtil.getValue("default", listener, IMessageConsumerListener::group);
        // 获取group中的pending消息信息，本质上就是执行XPENDING指令
        PendingMessagesSummary pendingMessagesSummary = streamOperations.pending(streamKey, group);
        if (null == pendingMessagesSummary) {
            return;
        }
        // 所有pending消息的数量
        long totalPendingMessages = pendingMessagesSummary.getTotalPendingMessages();
        if(0 == totalPendingMessages){
            return;
        }
        // 消费组名称
        String groupName= pendingMessagesSummary.getGroupName();
        // pending队列中的最小ID
        String minMessageId = pendingMessagesSummary.minMessageId();
        // pending队列中的最大ID
        String maxMessageId = pendingMessagesSummary.maxMessageId();
        logger.info("流：{},消费组：{}，一共有{}条pending消息，最大ID={}，最小ID={}", streamKey, groupName, totalPendingMessages, minMessageId, maxMessageId);
        // 获取每个消费者的pending消息数量:key消费者，value消息数量
        Map pendingMessageMap = pendingMessagesSummary.getPendingMessagesPerConsumer();
        if (pendingMessageMap.isEmpty()) {
            return;
        }

        Range range = Range.closed("0", "+");
        // 遍历每个消费者中的pending消息
        pendingMessageMap.forEach((key, totalMessages) -> {
            // 消费者的pending消息数量
            logger.info("消费者：{}，一共有{}条pending消息", key, totalMessages);
            if (totalMessages <= 0) {
                return;
            }
            // 读取消费者pending队列的前10条记录，从ID=0的记录开始，一直到ID最大值
            Consumer consumer = Consumer.from(groupName, key);
            PendingMessages pendingMessages = streamOperations.pending(streamKey, consumer, range, totalMessages);
            // 遍历所有Opending消息的详情
            pendingMessages.forEach(message -> {
                // 消息的ID
                RecordId recordId = message.getId();
                long deliveryCount = message.getTotalDeliveryCount();
                //对pending中未ack的数据进行重复消费
                if (deliveryCount > 0) {
                    String mid = recordId.getValue();
                    List> records = streamOperations.range(String.class, streamKey, Range.closed(mid, mid));
                    if (!CollectionUtils.isEmpty(records)) {
                        logger.info("pending 消息处理:records.size=" + records.size());
                        records.stream().filter(Objects::nonNull).forEach(this::onMessage);
                    }
                }
                // 消息从消费组中获取，到此刻的时间
                Duration elapsedTimeSinceLastDelivery = message.getElapsedTimeSinceLastDelivery();
                //判断是否为死信，如果是则ack掉
                if (elapsedTimeSinceLastDelivery.getSeconds() > DEATH_LINE) {
                    streamOperations.acknowledge(streamKey, groupName, recordId);
                }
            });
        });
    }

}

package io.shulie.jmeter.tool.redis.util;

import io.shulie.jmeter.tool.redis.RedisConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;


public class RedisUtil {
    private static final Logger logger = LoggerFactory.getLogger(RedisUtil.class);
    private final static Map jedisConnectionFactoryMap = new HashMap<>();
    private final static Map stringRedisTemplateMap = new HashMap<>();

    public static void checkAndCreateStreamGroup(StringRedisTemplate template, String streamKey, String group) {
        StreamInfo.XInfoGroups groups = null;
        try {
            groups = template.opsForStream().groups(streamKey);
        } catch (Exception e) {
            logger.error("groups方法执行失败", e);
        }
        if (null == groups || groups.isEmpty() || groups.stream().noneMatch(g -> g.groupName().equals(group))) {
            template.opsForStream().createGroup(streamKey, group);
        }
    }

    public static JedisConnectionFactory getJedisConnectionFactory(RedisConfig config) {
        JedisConnectionFactory factory = jedisConnectionFactoryMap.get(config);
        if (null == factory) {
            factory = buildJedisConnectionFactory(config);
            if (null != factory) {
                jedisConnectionFactoryMap.put(config, factory);
            }
        }
        return factory;
    }

    public static StringRedisTemplate getStringRedisTemplate(RedisConfig config) {
        JedisConnectionFactory factory = getJedisConnectionFactory(config);
        return getStringRedisTemplate(factory);
    }

    public static StringRedisTemplate getStringRedisTemplate(RedisConnectionFactory factory) {
        if (null == factory) {
            return null;
        }
        StringRedisTemplate template = stringRedisTemplateMap.get(factory);
        if (null == template) {
            template = buildStringRedisTemplate(factory);
            if (null != template) {
                stringRedisTemplateMap.put(factory, template);
            }
        }
        return template;
    }

    private static JedisConnectionFactory buildJedisConnectionFactory(RedisConfig config) {
        if (null == config) {
            return null;
        }
        JedisConnectionFactory factory;
        if (StringUtils.isNotBlank(config.getClusterNodes())) {
            factory = buildClusterJedisConnectionFactory(config);
        } else if (StringUtils.isNotBlank(config.getNodes())) {
            factory = buildSentineJedisConnectionFactory(config);
        } else {
            factory = buildStandaloneJedisConnectionFactory(config);
        }
        return factory;
    }

    private static StringRedisTemplate buildStringRedisTemplate(RedisConnectionFactory factory) {
        if (null == factory) {
            return null;
        }
        StringRedisTemplate template = new StringRedisTemplate();
        template.setDefaultSerializer(new StringRedisSerializer());
        template.setConnectionFactory(factory);
        template.afterPropertiesSet();
        return template;
    }

    
    private static JedisConnectionFactory buildSentineJedisConnectionFactory(RedisConfig redisConfig) {
        if (StringUtils.isBlank(redisConfig.getNodes())) {
            return null;
        }
        RedisSentinelConfiguration configuration = new RedisSentinelConfiguration();
        configuration.setMaster(redisConfig.getMaster());
        List redisNodes = Stream.of(redisConfig.getNodes()).filter(StringUtils::isNotBlank)
                .map(s -> s.split(","))
                .flatMap(Arrays::stream)
                .filter(StringUtils::isNotBlank)
                .map(s -> s.split(":"))
                .map(a -> new RedisNode(a[0], NumberUtils.toInt(a[1], 6379)))
                .distinct()
                .collect(Collectors.toList());
        configuration.setSentinels(redisNodes);
        if (StringUtils.isNotBlank(redisConfig.getPassword())) {
            configuration.setPassword(redisConfig.getPassword());
        }
        if (null != redisConfig.getDatabase()) {
            configuration.setDatabase(redisConfig.getDatabase());
        }

//        JedisClientConfiguration clientConfiguration = buildJedisClientConfiguration(redisConfig);
//        JedisPoolConfig config = buildJedisPoolConfig(redisConfig);

        JedisConnectionFactory factory = new JedisConnectionFactory(configuration);
//        if (null != clientConfiguration) {
//            factory = new JedisConnectionFactory(configuration, clientConfiguration);
//        } else {
//            factory = new JedisConnectionFactory(configuration);
//        }
//        if (null != config) {
//            factory.setPoolConfig(config);
//        }
        factory.afterPropertiesSet();
        return factory;
    }

    
    private static JedisConnectionFactory buildClusterJedisConnectionFactory(RedisConfig redisConfig) {
        if (StringUtils.isBlank(redisConfig.getClusterNodes())) {
            return null;
        }
        List redisNodes = Stream.of(redisConfig.getClusterNodes()).filter(StringUtils::isNotBlank)
                .map(s -> s.split(","))
                .flatMap(Arrays::stream)
                .filter(StringUtils::isNotBlank)
                .map(s -> s.split(":"))
                .map(a -> new RedisNode(a[0], NumberUtils.toInt(a[1], 6379)))
                .distinct()
                .collect(Collectors.toList());
        RedisClusterConfiguration configuration = new RedisClusterConfiguration();
        configuration.setClusterNodes(redisNodes);
        if (StringUtils.isNotBlank(redisConfig.getPassword())) {
            configuration.setPassword(redisConfig.getPassword());
        }

//        JedisClientConfiguration clientConfiguration = buildJedisClientConfiguration(redisConfig);
//        JedisPoolConfig config = buildJedisPoolConfig(redisConfig);

        JedisConnectionFactory factory = new JedisConnectionFactory(configuration);
//        if (null != clientConfiguration) {
//            factory = new JedisConnectionFactory(configuration, clientConfiguration);
//        } else {
//            factory = new JedisConnectionFactory(configuration);
//        }
//        if (null != config) {
//            factory.setPoolConfig(config);
//        }
        factory.afterPropertiesSet();
        return factory;
    }

    
    private static JedisConnectionFactory buildStandaloneJedisConnectionFactory(RedisConfig redisConfig) {
        //单机模式
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        if (StringUtils.isNotBlank(redisConfig.getHost())) {
            configuration.setHostName(redisConfig.getHost());
        }
        if (null != redisConfig.getPort()) {
            configuration.setPort(redisConfig.getPort());
        }
        if (StringUtils.isNotBlank(redisConfig.getPassword())) {
            configuration.setPassword(redisConfig.getPassword());
        }
        if (null != redisConfig.getDatabase()) {
            configuration.setDatabase(redisConfig.getDatabase());
        }

        JedisConnectionFactory factory = new JedisConnectionFactory(configuration);
        factory.afterPropertiesSet();
        return factory;
    }

//    private static JedisClientConfiguration buildJedisClientConfiguration(RedisConfig redisConfig) {
//        JedisClientConfiguration.JedisClientConfigurationBuilder builder = null;
//        if (null != redisConfig.getTimeout()) {
//            builder = JedisClientConfiguration.builder();
//            builder.connectTimeout(Duration.ofSeconds(redisConfig.getTimeout()));
//        }
//        if (null != redisConfig.getSoTimeOut()) {
//            if (null == builder) {
//                builder = JedisClientConfiguration.builder();
//            }
//            builder.readTimeout(Duration.ofSeconds(redisConfig.getSoTimeOut()));
//        }
//        if (null != builder) {
//            return builder.build();
//        }
//        return null;
//    }

//    private static JedisPoolConfig buildJedisPoolConfig(RedisConfig redisConfig) {
//        JedisPoolConfig config = null;
//        if (null != redisConfig.getMaxTotal()) {
//            config = new JedisPoolConfig();
//            config.setMaxTotal(redisConfig.getMaxTotal());
//        }
//        if (null != redisConfig.getMaxIdle()) {
//            if (null == config) {
//                config = new JedisPoolConfig();
//            }
//            config.setMaxIdle(redisConfig.getMaxIdle());
//        }
//        return config;
//    }


}

public class RedisConfig {
    private String nodes;
    private String master;
    private String host;
    private Integer port;
    private Integer database;
    
    private String password;
    private String clusterNodes;

    private Integer maxIdle;
    private Integer maxTotal;
    private Integer timeout;
    private Integer soTimeOut;
}

     *
     * @param defValue 默认值
     * @param t        取值对象
     * @param func     取值方法，从对象中取值的方法
     * @param       取值对象类型
     * @param       返回值对象类型
     */
    public static  R getValue(R defValue, T t, Function func) {
        R result = defValue;
        if (null != t) {
            R r = func.apply(t);
            if (null != r) {
                if (r instanceof String) {
                    if (StringUtils.isNotBlank((String)r)) {
                        result = r;
                    }
                } else if (r instanceof List) {
                    if (!CollectionUtils.isEmpty((List)r)) {
                        result = r;
                    }
                } else if (r instanceof Map) {
                    if (((Map) r).isEmpty()) {
                        result = r;
                    }
                } else {
                    result = r;
                }
            }
        }
        return result;
    }

    public static boolean matchTags(IMessageConsumerListener listener, TkMessage message) {
        if (StringUtils.isBlank(message.getTag())) {
            return true;
        }
        List tags = listener.getTags();
        if (CollectionUtils.isEmpty(tags)) {
            return false;
        }
        try {
            return tags.stream().filter(Objects::nonNull).anyMatch(s -> matches(message.getTag(), s));
        } catch (Exception e) {
            log.error("", e);
        }
        return false;
    }

    private static boolean matches(String s, String match) {
        if (s.equals(match)) {
            return true;
        }
        if ("*".equals(match)) {
            return true;
        }
        if (match.startsWith("*") && match.endsWith("*")) {
            match = match.substring(1, match.length()-1);
            return s.contains(match);
        } else if (match.startsWith("*")) {
            match = match.substring(1);
            return s.endsWith(match);
        } else if (match.endsWith("*")) {
            match = match.substring(0, match.length()-1);
            return s.startsWith(match);
        }
        return false;
    }
}

public class JsonUtil {
    private static final Logger log = LoggerFactory.getLogger(JsonUtil.class);

    public static String toJson(Object o) {
        if (null == o) {
            return null;
        }
        try {
            return JSON.toJSONString(o);
        } catch (Exception e) {
            log.error("toJson failed!o="+o);
        }
        return null;
    }

    public static JSONObject parse(String text) {
        if (StringUtils.isBlank(text)) {
            return null;
        }
        JSONObject json = null;
        try {
            json = JSON.parseObject(text);
        } catch (Exception e) {
            log.error("parse json failed!text="+text);
        }
        return json;
    }

    public static  T parseObject(String text, Class clazz) {
        if (StringUtils.isBlank(text)) {
            return null;
        }
        T result = null;
        try {
            result = JSON.parseObject(text, clazz);
        } catch (Exception e) {
            log.error("parse json to object class failed!text="+text);
        }
        return result;
    }

    public static  T parseObject(String text, TypeReference type) {
        if (StringUtils.isBlank(text)) {
            return null;
        }
        T result = null;
        try {
            result = JSON.parseObject(text, type);
        } catch (Exception e) {
            log.error("parse json to object type failed!text="+text);
        }
        return result;
    }

    public static JSONArray parseArray(String text) {
        if (StringUtils.isBlank(text)) {
            return null;
        }
        JSONArray result = null;
        try {
            result = JSON.parseArray(text);
        } catch (Exception e) {
            log.error("json parseArray failed!text="+text);
        }
        return result;
    }

    public static  List parseArray(String text, Class clazz) {
        if (StringUtils.isBlank(text)) {
            return null;
        }
        List result = null;
        try {
            result = JSON.parseArray(text, clazz);
        } catch (Exception e) {
            log.error("json parseArray failed!text="+text);
        }
        return result;
    }


}

@Slf4j
@Service
public class MessageConsumerService extends AbstractMessageConsumerListener {
    public static final GroupTopicEnum groupTopic = new GroupTopicEnum("yanbatest3", "default-test");
//    @Resource
//    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedisConfig redisConfig;
    private MessageConsumer messageConsumer;

    @PostConstruct
    private void init() {
        messageConsumer = MessageConsumer.getInstance(redisConfig);
        messageConsumer.setThreadPool(Executors.threadPool);
        messageConsumer.register(this);
    }

    @Override
    public GroupTopicEnum getGroupTopic() {
        return groupTopic;
    }

    @Override
    public List getTags() {
        return Lists.newArrayList("*");
    }

    @Override
    public boolean receive(TkMessage message) {
        log.info("message="+ JsonUtil.toJson(message));
        return true;
    }

}

@Slf4j
@Service
public class MessageProducerService {
//    @Resource
//    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedisConfig redisConfig;
    private MessageProducer messageProducer;
    @PostConstruct
    public void init() {
        messageProducer = MessageProducer.getInstance(redisConfig);
    }

    public boolean send(String key, String content) {
        TkMessage message = TkMessage.create()
                .setGroupTopic(MessageConsumerService.groupTopic)
                .setTag("tagA")
                .setKey(key)
                .setContent(content)
                .build();
        return send(message);
    }

    public boolean send(TkMessage message) {
        try {
            return messageProducer.send(message);
        } catch (Exception e) {
            log.error("send message error", e);
        }
        return false;
    }
}

这里要特别说明一下，消费者IMessageConsumerListener定义多个方法，其中group方法表示消费组，

        如果多个消费者返回的值相同表示消费者共同竞争，只会有一个消费者收到消息，即消息单点消费；

        如果多个消费者返回的值不同表示消费者都需要收到消息，即消息广播，所有消费者都进行消费；