@Slf4j
@Service
public class ListRedisQueue {

    //隊(duì)列名
    public static final String KEY = "listQueue";

    @Resource
    private RedisTemplate redisTemplate;

    public void produce(String message) {
        redisTemplate.opsForList().rightPush(KEY, message);
    }

    public void consume() {
        while (true) {
            String msg = (String) redisTemplate.opsForList().leftPop(KEY);
            log.info("瘋狂獲取消息：" + msg);
        }
    }

    public void blockingConsume() {
        while (true) {
            List<Object> obj = redisTemplate.executePipelined(new RedisCallback<Object>() {

                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    //隊(duì)列沒(méi)有元素會(huì)阻塞操作，直到隊(duì)列獲取新的元素或超時(shí)，5表示如果沒(méi)元素就每五秒去拿一次消息
                    return connection.bRPop(5, KEY.getBytes());
                }
            }, new StringRedisSerializer());
            for (Object str : obj) {
                log.info("blockingConsume獲取消息 : {}", str);
            }
        }
    }

}

    @Autowired
    private ListRedisQueue listRedisQueue;

    @Test
    public void produce() {
        for (int i = 0; i < 5; i++) {
            listRedisQueue.produce("第"+i + "個(gè)數(shù)據(jù)");
        }
    }

    @Test
    public void consume() {
        produce();
        logger.info("生產(chǎn)消息完畢");
        listRedisQueue.consume();
    }

    @Test
    public void blockingConsume() {
        produce();
        logger.info("生產(chǎn)消息完畢");
        listRedisQueue.blockingConsume();
    }

@Slf4j
@Service
public class SortedSetRedisQueue {
    //隊(duì)列名
    public static final String KEY = "sortedSet_queue";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void produce(String msg, Double score) {
        // 創(chuàng)建Sorted Set實(shí)例
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
        // 添加數(shù)據(jù)
        zSetOperations.add(KEY, msg, score);
    }

    public void consumer() throws InterruptedException {
        // 創(chuàng)建SortedSet實(shí)例
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
        while (true) {
            // 拿取數(shù)據(jù) (rangeByScore返回有序集合中指定分?jǐn)?shù)區(qū)間的成員列表。有序集成員按分?jǐn)?shù)值遞增(從小到大)次序排列)
            Set<String> order = zSetOperations.rangeByScore(KEY, 0, System.currentTimeMillis(), 0, 1);
            if (ObjectUtils.isEmpty(order)) {
                log.info("當(dāng)前沒(méi)有數(shù)據(jù) 當(dāng)前線程睡眠3秒");
                TimeUnit.SECONDS.sleep(3);
                // 跳過(guò)本次循環(huán) 重新循環(huán)拿取數(shù)據(jù)
                continue;
            }
            // 利用迭代器拿取Set中的數(shù)據(jù)
            String massage = order.iterator().next();
            // 過(guò)河拆遷，拿到就刪除消息
            if (zSetOperations.remove(KEY, massage) > 0) {
                //做些業(yè)務(wù)處理
                log.info("我拿到的消息：" + massage);
            }
        }

    }
}

    @Autowired
    private SortedSetRedisQueue sortedSetRedisQueue;
    @Test
    public void sortedSetProduce() throws InterruptedException {
        for (int i = 0; i < 5; i++) {
            TimeUnit.SECONDS.sleep(1);
            // 生成分?jǐn)?shù)
            double score = System.currentTimeMillis();
            sortedSetRedisQueue.produce("第"+i + "個(gè)數(shù)據(jù)",score);
        }
    }
    @Test
    public void sortedSetConsumer() throws InterruptedException {
        sortedSetProduce();
        logger.info("生產(chǎn)消息完畢");
        sortedSetRedisQueue.consumer();
    }
}

@Slf4j
@Component
public class RedisMessageListenerListener implements MessageListener {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 消息處理
     *
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = new String(pattern);
        log.info("onMessage --> 消息通道是：{}", channel);

        RedisSerializer<?> valueSerializer = redisTemplate.getValueSerializer();
        Object deserialize = valueSerializer.deserialize(message.getBody());
        log.info("反序列化的結(jié)果:{}", deserialize);
        if (deserialize == null) return;
        String md5DigestAsHex = DigestUtils.md5DigestAsHex(deserialize.toString().getBytes(StandardCharsets.UTF_8));
        log.info("計(jì)算得到的key: {}", md5DigestAsHex);
        Boolean result = redisTemplate.opsForValue().setIfAbsent(md5DigestAsHex, "1", 20, TimeUnit.SECONDS);
        if (Boolean.TRUE.equals(result)) {
            // redis消息進(jìn)行處理
            log.info("接收的結(jié)果：{}", deserialize.toString());
        } else {
            log.info("其他服務(wù)處理中");
        }
    }
}

@Configuration
public class RedisMessageListenerConfig {
    @Bean
    public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,
                                                                          RedisMessageListenerListener redisMessageListenerListener) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
        redisMessageListenerContainer.addMessageListener(redisMessageListenerListener, new ChannelTopic(PubSubRedisQueue.KEY));
        return redisMessageListenerContainer;
    }
}

@Service
public class PubSubRedisQueue {
    //隊(duì)列名
    public static final String KEY = "pub_sub_queue";

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void produce(String message) {
        redisTemplate.convertAndSend(KEY, message);
    }
}  

@Slf4j
@RestController
@RequestMapping(value = "/queue")
public class RedisMQController {
    @Autowired
    private PubSubRedisQueue pubSubRedisQueue;

    @RequestMapping(value = "/pubsub/produce", method = RequestMethod.GET)
    public void pubsubProduce(@RequestParam(name = "msg") String msg) {
        pubSubRedisQueue.produce(msg);
    }

@Slf4j
@Component
public class AutoAckStreamConsumeListener implements StreamListener<String, MapRecord<String, String, String>> {
    //分組名
    public static final String GROUP = "autoack_stream";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        Map<String, String> map = message.getValue();
        log.info("[自動(dòng)ACK]接收到一個(gè)消息 stream:[{}],id:[{}],value:[{}]", stream, id, map);
        redisTemplate.opsForStream().delete(GROUP, id.getValue());
    }
}

@Slf4j
@Component
public class BasicAckStreamConsumeListener implements StreamListener<String, MapRecord<String, String, String>> {
    //分組名
    public static final String GROUP = "basicack_stream";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        Map<String, String> map = message.getValue();
        log.info("[手動(dòng)ACK]接收到一個(gè)消息 stream:[{}],id:[{}],value:[{}]", stream, id, map);
        redisTemplate.opsForStream().acknowledge(stream, GROUP, id.getValue());
        //消費(fèi)完畢刪除該條消息
        redisTemplate.opsForStream().delete(GROUP, id.getValue());
    }
}

@Slf4j
@Configuration
public class RedisStreamConfiguration {
    @Autowired
    private RedisConnectionFactory redisConnectionFactory;
    @Autowired
    private AutoAckStreamConsumeListener autoAckStreamConsumeListener;
    @Autowired
    private BasicAckStreamConsumeListener basicAckStreamConsumeListener;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer() {
        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(3)
                        // 運(yùn)行 Stream 的 poll task
                        .executor(executor)
                        // Stream 中沒(méi)有消息時(shí)，阻塞多長(zhǎng)時(shí)間，需要比 `spring.redis.timeout` 的時(shí)間小
                        .pollTimeout(Duration.ofSeconds(3))
                        // 獲取消息的過(guò)程或獲取到消息給具體的消息者處理的過(guò)程中，發(fā)生了異常的處理
                        .errorHandler(new ErrorHandler() {
                            @Override
                            public void handleError(Throwable t) {
                                log.info("出現(xiàn)異常就來(lái)這里了" + t);
                            }
                        })
                        .build();

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);
        // 獨(dú)立消費(fèi)

        // 消費(fèi)組A,自動(dòng)ack
        // 從消費(fèi)組中沒(méi)有分配給消費(fèi)者的消息開(kāi)始消費(fèi)
        if (!isStreamGroupExists(StreamRedisQueue.KEY,AutoAckStreamConsumeListener.GROUP)){
            redisTemplate.opsForStream().createGroup(StreamRedisQueue.KEY,AutoAckStreamConsumeListener.GROUP);
        }
        streamMessageListenerContainer.receiveAutoAck(Consumer.from(AutoAckStreamConsumeListener.GROUP, "AutoAckConsumer"),
                StreamOffset.create(StreamRedisQueue.KEY, ReadOffset.lastConsumed()), autoAckStreamConsumeListener);

        // 消費(fèi)組B,不自動(dòng)ack
        if (!isStreamGroupExists(StreamRedisQueue.KEY,BasicAckStreamConsumeListener.GROUP)){
            redisTemplate.opsForStream().createGroup(StreamRedisQueue.KEY,BasicAckStreamConsumeListener.GROUP);
        }
        streamMessageListenerContainer.receive(Consumer.from(BasicAckStreamConsumeListener.GROUP, "BasicAckConsumer"),
                StreamOffset.create(StreamRedisQueue.KEY, ReadOffset.lastConsumed()), basicAckStreamConsumeListener);

        return streamMessageListenerContainer;

    }

    /**
     * 判斷該消費(fèi)組是否存在
     * @param streamKey
     * @param groupName
     * @return
     */
    public boolean isStreamGroupExists(String streamKey, String groupName) {
        RedisStreamCommands commands = redisConnectionFactory.getConnection().streamCommands();
        //首先檢查Stream Key是否存在，否則下面代碼可能會(huì)因?yàn)閲L試檢查不存在的Stream Key而導(dǎo)致異常
        if (!redisTemplate.hasKey(streamKey)){
            return false;
        }
        //獲取streamKey下的所有g(shù)roups
        StreamInfo.XInfoGroups xInfoGroups = commands.xInfoGroups(streamKey.getBytes());
        AtomicBoolean exists= new AtomicBoolean(false);
        xInfoGroups.forEach(xInfoGroup -> {
            if (xInfoGroup.groupName().equals(groupName)){
                exists.set(true);
            }
        });

        return exists.get();
    }
}

@Slf4j
@Service
public class StreamRedisQueue {
    //隊(duì)列名
    public static final String KEY = "stream_queue";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public String  produce(Map<String, String> value) {
        return redisTemplate.opsForStream().add(KEY, value).getValue();
    }

    public void createGroup(String key, String group){
        redisTemplate.opsForStream().createGroup(key, group);
    }

}

@Slf4j
@RestController
@RequestMapping(value = "/queue")
public class RedisMQController {

    @Autowired
    private StreamRedisQueue streamRedisQueue;
    @RequestMapping(value = "/stream/produce", method = RequestMethod.GET)
    public void streamProduce() {
        Map<String, String> map = new HashMap<>();
        map.put("劉德華", "大家好我是劉德華");
        map.put("周杰倫", "周杰倫");
        map.put("time", DateUtil.now());
        String result = streamRedisQueue.produce(map);
        log.info("返回結(jié)果：{}", result);
    }
}