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

spring:
  data:
    redis:
    # 改為自己的地址和密碼
      host: 10.69.37.213
      port: 6379
      password: Jolly
      # 連接超時(shí)時(shí)間,單位ms
      connect-timeout: 50000
      # 選擇第幾個(gè)數(shù)據(jù)庫，默認(rèn)為0,最大值15
      database: 0
      lettuce:
        pool:
          # 最大的活躍連接數(shù)，不會(huì)超過這個(gè)數(shù)，根據(jù)項(xiàng)目預(yù)期并發(fā)量調(diào)整
          max-active: 50
          # max-idle 指定了連接池中的最大空閑連接數(shù)。
          # 空閑連接是指那些沒有被使用，但是已經(jīng)創(chuàng)建并且保持在連接池中的連接
          # 這個(gè)值應(yīng)該與max-active相匹配或者稍微低一些，
          # 以保持連接池中有足夠的空閑連接來處理突發(fā)請(qǐng)求。
          # 設(shè)置得太高可能會(huì)導(dǎo)致資源浪費(fèi)，因?yàn)榭臻e連接會(huì)占用內(nèi)存和其他資源。
          max-idle: 30
          #這個(gè)配置指定了連接池中的最小空閑連接數(shù)。
          #這個(gè)設(shè)置可以確保連接池始終保持一定數(shù)量的空閑連接，以便在請(qǐng)求到來時(shí)可以立即使用，而不需要等待連接的創(chuàng)建。
          # 這個(gè)值應(yīng)該根據(jù)你的應(yīng)用程序的基線負(fù)載來設(shè)置
          min-idle: 10
          # 當(dāng)連接池達(dá)到最大活躍連接數(shù)時(shí)，客戶端等待可用連接的最大時(shí)間（以毫秒為單位）。-1 表示無限等待
          # 如果設(shè)置為一個(gè)正數(shù)，那么在等待時(shí)間超過這個(gè)值后，會(huì)拋出一個(gè)異常。
          max-wait: -1

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;

@Configuration
public class redisConfig {

	@ConditionalOnMissingBean(name = "redisTemplate")
    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        // 默認(rèn)為utf-8,可以進(jìn)行修改
        template.setKeySerializer(new StringRedisSerializer());
        // 原版默認(rèn)使用jdk的序列化方式JdkSerializationRedisSerializer
        Jackson2JsonRedisSerializer serializer = new Jackson2JsonRedisSerializer(Object.class);
        template.setValueSerializer(serializer);
        // 設(shè)置Hash的序列化化方式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(serializer);
        // 設(shè)置屬性
        template.afterPropertiesSet();
        return template;
    }
}

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class order implements Serializable {
    private int id;
    private String userid;
    private String goodName;
}

import jakarta.annotation.Resource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class Mq {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

	// 隊(duì)列名
    private final String queue = "order_queue";

    @GetMapping("/order")
    public void order(){
       	// 為了模擬消息的獲取，異步開啟一個(gè)線程，進(jìn)行消息處理
        Thread thread = new Thread(() -> {
            try {
                processData();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        thread.start();
         // 模擬產(chǎn)生10條消息，放入隊(duì)列
        for (int i = 0; i < 10; i++) {
            order order = new order(i, i, "goods" + i);
            redisTemplate.opsForList().leftPush(queue, order);
            System.out.println("放入消息隊(duì)列："+i);
        }
    }

    // 處理消息，不斷的輪詢隊(duì)列中的消息
    public void processData() throws InterruptedException {
        while (true){
            Object order = redisTemplate.opsForList().rightPop(queue);
            if(order == null){
                System.out.println("當(dāng)前沒有消息");
                Thread.sleep(1000);
            }else{
                System.out.println("處理消息："+order);
            }
        }
    }
}

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private final String CHANNEL_NAME  = "order_queue";
    
	@GetMapping("/order")
    public void order(){
        // 模擬產(chǎn)生10條消息，放入隊(duì)列
        for (int i = 0; i < 10; i++) {
            order order = new order(i, i, "goods" + i);
            //發(fā)布消息
            redisTemplate.convertAndSend(CHANNEL_NAME, order);
            System.out.println("放入消息隊(duì)列："+i);
        }
    }

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;

@Service
public class sub implements MessageListener {
	
	// 當(dāng)監(jiān)聽到有消息的時(shí)候，就會(huì)執(zhí)行這個(gè)方法
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String msg = new String(message.getBody());
        // 模擬延遲處理
        try {
            Thread.sleep(2000);  // 假設(shè)處理需要2秒
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("處理消息："+msg);
    }

}

	/**
     * 因?yàn)闃?biāo)記了@Bean 注解，所以會(huì)在springboot啟動(dòng)的時(shí)候調(diào)用該方法創(chuàng)建，也可以放在其他地方進(jìn)行創(chuàng)建
     * 當(dāng)調(diào)用這個(gè)方法時(shí)，RedisConnectionFactory 這個(gè)對(duì)象已經(jīng)存在于springboot的容器內(nèi)，然后調(diào)用這個(gè)
     * 方法的時(shí)候就會(huì)傳入該參數(shù)，執(zhí)行方法后會(huì)創(chuàng)建一個(gè)RedisMessageListenerContainer，這樣可以在其他類
     * 里面管理這些監(jiān)聽MessageListener
     * @param connectionFactory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer redisContainer(RedisConnectionFactory connectionFactory) {
        // 首先創(chuàng)建一個(gè)監(jiān)聽的容器，這個(gè)容器可以傳入多個(gè)MessageListener
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        // 注入一個(gè)連接池工廠
        container.setConnectionFactory(connectionFactory);
        // 創(chuàng)建一個(gè)自己的監(jiān)聽類
        sub sub = new sub();
        // 然后和名為order_queue的通道進(jìn)行綁定
        container.addMessageListener(sub,new ChannelTopic("order_queue"));
        return container;
    }

import com.xujialin.springboot3_study.entity.order;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;

@RestController
public class streamCont {

    @Resource
    private RedisTemplate<String,Object> redisTemplate;

    private String stream_key = "stream_key";



    @GetMapping("/order")
    public void order() {
        //封裝請(qǐng)求,假裝這是高并發(fā)場(chǎng)景
        for (int j = 0; j < 100; j++) {
            new Thread(() -> {
                for (int i = 0; i < 100; i++) {
                    order order = new order(i, i, "goods" + i);
                    HashMap<String,order> map = new HashMap<>();
                    map.put("order", order);
                    Long size = redisTemplate.opsForStream().size(stream_key);
                    if(size > 500){
                        System.out.println("活動(dòng)太火爆了，請(qǐng)重試！");
                        continue;
                    }
                    redisTemplate.opsForStream().add(stream_key,map);
                }
            }).start();
        }
        System.out.println("恭喜你搶到了");
    }

    @PostConstruct
    public void init(){
        // 第一個(gè)是stream的key,第二個(gè)是組名
//        redisTemplate.opsForStream().createGroup(stream_key, "g1");
        try {
            redisTemplate.opsForStream().createGroup(stream_key, "g1");
        } catch (RedisSystemException e) {
            // 如果 group 已存在，拋出異常，可忽略
            System.out.println("group已經(jīng)存在");
        }
        for (int i = 0; i < 5; i++) {
            new Thread(new consumer()).start();
        }
    }

    class consumer implements Runnable {

        @Override
        public void run() {
            while(true){
                // 讀取消息
                List<MapRecord<String, Object, Object>> list = redisTemplate.opsForStream().read(
                        // 為g1消費(fèi)者組創(chuàng)建一個(gè)消費(fèi)則名字，可以為每個(gè)線程的名字，也可以為一個(gè)固定的名字，
                        // 一條消息最多只能被組里面的一個(gè)消費(fèi)者消費(fèi)，如果一條消息同時(shí)被兩個(gè)消費(fèi)者消費(fèi)，
                        // 那么這兩個(gè)消費(fèi)者應(yīng)該隸屬于不同的消費(fèi)者組，所以同一個(gè)名字或者不同的名字，對(duì)于同一個(gè)
                        // 消費(fèi)組沒有太大區(qū)別
                        Consumer.from("g1", Thread.currentThread().getName()),
                        // 創(chuàng)建一個(gè)讀取選項(xiàng)，創(chuàng)建一個(gè)空的 StreamReadOptions 實(shí)例。這是配置讀取選項(xiàng)的起點(diǎn)
                        // .count(1): 設(shè)置讀取操作返回的最大消息數(shù)量。意味著每次讀取操作最多只會(huì)返回一條消息。
                        //.block(Duration.ofSeconds(2)): 配置讀取操作為阻塞模式，并設(shè)置阻塞的超時(shí)時(shí)間為2s,
                        // 也可以設(shè)置單位
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        // 創(chuàng)建一個(gè)偏移量，ReadOffset.lastConsumed(): 這是指定讀取消息的偏移量。
                        // 表示從消費(fèi)者組中最后一次確認(rèn)消費(fèi)的消息之后開始讀取新的消息。
                        StreamOffset.create( "stream_key", ReadOffset.lastConsumed()));

                // 讀取消息
                if (list != null && !list.isEmpty()) {
                    MapRecord<String, Object, Object> entries = list.get(0);
                    // 模擬處理消息
                    System.out.println(entries);
                    // 確認(rèn)消息
                    redisTemplate.opsForStream().acknowledge("stream_key","g1",entries.getId());
                }
            }
        }
    }
}