/**
 * Redis Stream 消息隊(duì)列配置
 */
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    private static final Logger log = LoggerFactory.getLogger(RedisStreamConfiguration.class);
    private final RedisConnectionFactory redisConnectionFactory;
    private final Consumer1 Consumer1;
    private final Consumer2 Consumer2;

    // 定義需要自定義的配置常量
    private static final int BATCH_SIZE = 10; // 每次批量拉取的消息數(shù)量
    private static final Duration POLL_TIMEOUT = Duration.ofSeconds(3); // 拉取消息的阻塞超時(shí)時(shí)間
    private static final String THREAD_NAME_PREFIX = "your-business"; // 線程名稱前綴
    private static final String GROUP_NAME_1 = "group1"; // 第一個(gè)消費(fèi)者組名稱
    private static final String GROUP_NAME_2 = "group2"; // 第二個(gè)消費(fèi)者組名稱
    private static final String CONSUMER_NAME_1 = "consumer1"; // 第一個(gè)消費(fèi)者名稱
    private static final String CONSUMER_NAME_2 = "consumer2"; // 第二個(gè)消費(fèi)者名稱
    private static final String STREAM_TOPIC_KEY = SHORT_LINK_STATS_STREAM_TOPIC_KEY; // Stream的主題鍵

    @Bean
    public ExecutorService asyncStreamConsumer() {
        log.info("Redis Stream 消息隊(duì)列配置線程池");
        AtomicInteger index = new AtomicInteger();
        int processors = Runtime.getRuntime().availableProcessors();

        // 創(chuàng)建一個(gè)自定義線程池
        return new ThreadPoolExecutor(
                processors,
                processors + (processors >> 1),
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                runnable -> {
                    Thread thread = new Thread(runnable);
                    thread.setName(THREAD_NAME_PREFIX + "_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(
            ExecutorService asyncStreamConsumer) {

        // 配置 StreamMessageListenerContainer 容器選項(xiàng)
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .batchSize(BATCH_SIZE) // 批量拉取消息數(shù)量
                        .executor(asyncStreamConsumer) // 使用配置好的線程池
                        .pollTimeout(POLL_TIMEOUT) // 拉取消息的超時(shí)時(shí)間
                        .build();

        // 創(chuàng)建 StreamMessageListenerContainer 實(shí)例
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 配置第一個(gè)消息監(jiān)聽器
        container.receiveAutoAck(
                Consumer.from(GROUP_NAME_1, CONSUMER_NAME_1), // 指定第一個(gè)消費(fèi)者組和消費(fèi)者名稱
                StreamOffset.create(STREAM_TOPIC_KEY, ReadOffset.lastConsumed()), // 指定主題和偏移量
                Consumer1 // 指定第一個(gè)消息處理邏輯
        );

        // 配置第二個(gè)消息監(jiān)聽器
        container.receiveAutoAck(
                Consumer.from(GROUP_NAME_2, CONSUMER_NAME_2), // 指定第二個(gè)消費(fèi)者組和消費(fèi)者名稱
                StreamOffset.create(STREAM_TOPIC_KEY, ReadOffset.lastConsumed()), // 指定主題和偏移量
                Consumer2 // 指定第二個(gè)消息處理邏輯
        );

        return container;
    }
}

@Bean
public ExecutorService asyncStreamConsumer() { ... }

@Bean(initMethod = "start", destroyMethod = "stop")
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(...) { ... }

/**
 * 消息隊(duì)列消費(fèi)者
 */
@RequiredArgsConstructor
@Slf4j
@Component
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {
   
    private final RedissonClient redissonClient;
    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;


    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        if (!messageQueueIdempotentHandler.isMessageProcessed(id.toString())) {
            // 判斷當(dāng)前的這個(gè)消息流程是否執(zhí)行完成
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息隊(duì)列重試");
        }
        try {
            Map<String, String> producerMap = message.getValue();
         	//你自己的業(yè)務(wù)邏輯
            }
            // 刪除消息
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex) {
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("消費(fèi)異常", ex);
            throw ex;
        }
        //消費(fèi)完刪除
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }

   
}

/**
 * 短鏈接監(jiān)控狀態(tài)保存消息隊(duì)列生產(chǎn)者
 */
@Component
@RequiredArgsConstructor
public class Producer implements MessageQueueProducer{

    private final StringRedisTemplate stringRedisTemplate;


    /**
     * 發(fā)送消息
     */
    public void send(Map<String, String> producerMap) {
        stringRedisTemplate.opsForStream().add(YOUR_KEY, producerMap);
    }
}