SpringBoot與Redis多線程入門——多線程redis存取數據

1. SpringBoot Redis yml 配置

此處省略密碼

spring:
  redis:
    database: 0
    host: 127.0.0.1
    port: 6379
    timeout: 3000
    jedis:
      pool:
        max-active: 8
        max-wait: -1
        max-idle: 8
        min-idle: 0

2. RedisCofig.java 配置類代碼

@EnableCaching
@Configuration
public class RedisConfig {

	@Value("${spring.redis.host}")
	private String host;
	@Value("${spring.redis.database}")
	private Integer database;
	@Value("${spring.redis.port}")
	private Integer port;

	@Primary
	@Bean(name = "jedisPoolConfig")
	@ConfigurationProperties(prefix = "spring.redis.pool")
	public JedisPoolConfig jedisPoolConfig() {
		JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
		jedisPoolConfig.setMaxWaitMillis(10000);
		return jedisPoolConfig;
	}

	@Bean
	public RedisConnectionFactory redisConnectionFactory(JedisPoolConfig jedisPoolConfig) {
		RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
		redisStandaloneConfiguration.setHostName(host);
		redisStandaloneConfiguration.setDatabase(database);
		// redisStandaloneConfiguration.setPassword(pwd);
		redisStandaloneConfiguration.setPort(port);
		JedisClientConfiguration.JedisPoolingClientConfigurationBuilder jpcb = (JedisClientConfiguration.JedisPoolingClientConfigurationBuilder) JedisClientConfiguration.builder();
		jpcb.poolConfig(jedisPoolConfig);
		JedisClientConfiguration jedisClientConfiguration = jpcb.build();
		return new JedisConnectionFactory(redisStandaloneConfiguration, jedisClientConfiguration);
	}

	/**
	 * 配置redisTemplate針對不同key和value場景下不同序列化的方式
	 * 此處針對key為String，value為CustomerVo對象的序列化方式
	 * @param factory Redis連接工廠
	 * @return
	 */
	@Primary
	@Bean(name = "customerRedisTemplate")
	public RedisTemplate<String, CustomerVo> customerRedisTemplate(RedisConnectionFactory factory) {
		RedisTemplate<String, CustomerVo> template = new RedisTemplate<>();
		template.setConnectionFactory(factory);
		StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
		template.setKeySerializer(stringRedisSerializer);
		template.setHashKeySerializer(stringRedisSerializer);
                // 這里是關鍵，注意替換為自己的類
		Jackson2JsonRedisSerializer<CustomerVo> redisSerializer = new Jackson2JsonRedisSerializer<>(CustomerVo.class);
		template.setValueSerializer(redisSerializer);
		template.setHashValueSerializer(redisSerializer);
		template.afterPropertiesSet();
		return template;
	}

	@Bean(name = "doctorRedisTemplate")
	public RedisTemplate<String, DoctorVo> doctorRedisTemplate(RedisConnectionFactory factory) {
		RedisTemplate<String, DoctorVo> template = new RedisTemplate<>();
		template.setConnectionFactory(factory);
		StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
		template.setKeySerializer(stringRedisSerializer);
		template.setHashKeySerializer(stringRedisSerializer);
		Jackson2JsonRedisSerializer<DoctorVo> redisSerializer = new Jackson2JsonRedisSerializer<>(DoctorVo.class);
		template.setValueSerializer(redisSerializer);
		template.setHashValueSerializer(redisSerializer);
		template.afterPropertiesSet();
		return template;
	}



	@Bean
	@ConditionalOnMissingBean
	@ConditionalOnSingleCandidate(RedisConnectionFactory.class)
	public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
		StringRedisTemplate template = new StringRedisTemplate();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}
}

3. Vo類和Service類代碼

CustomerVo.java

@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class CustomerVo {
	private Integer customerId;
	private String queueSeq;
	private String customerName;
	private String customerSex;

	@Override
	public String toString() {
		return "CustomerVo{" +
				"queueSeq='" + queueSeq + '\'' +
				", customerName='" + customerName + '\'' +
				", customerSex='" + customerSex + '\'' +
				'}';
	}
}

Service

@Slf4j
@Service
public class RedisLookupService {

	@Autowired
	private RedisTemplate<String, CustomerVo> redisTemplate;

	@Async("taskExecutor")
	public CompletableFuture<Long> enqueueCustomer(CustomerVo customer) {
		Long result = redisTemplate.opsForList().rightPush("queue", customer);
		log.info("{} 入隊..", customer);
		return CompletableFuture.completedFuture(result);
	}

	@Async("taskExecutor")
	public CompletableFuture<CustomerVo> dequeueCustomer() {
		if (Objects.requireNonNull(redisTemplate.opsForList().size("queue")) < 1) {
			return CompletableFuture.completedFuture(null);
		}
		CustomerVo vo = redisTemplate.opsForList().leftPop("queue");
		log.info("{} 出隊...", vo);
		return CompletableFuture.completedFuture(vo);
	}

}

AsyncConfig.java 配置類

因為用到了SpringBoot的多線程，所以要加一下這個配置類

@Configuration
@EnableAsync  // 啟用異步任務
public class AsyncConfig {

	// 聲明一個線程池(並指定線程池的名字)
	@Bean("taskExecutor")
	public Executor asyncExecutor() {
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		//核心線程數5：線程池創建時候初始化的線程數
		executor.setCorePoolSize(5);
		//最大線程數5：線程池最大的線程數，只有在緩沖隊列滿了之后才會申請超過核心線程數的線程
		executor.setMaxPoolSize(5);
		//緩沖隊列500：用來緩沖執行任務的隊列
		executor.setQueueCapacity(500);
		//允許線程的空閑時間60秒：當超過了核心線程出之外的線程在空閑時間到達之后會被銷毀
		executor.setKeepAliveSeconds(60);
		//線程池名的前綴：設置好了之后可以方便我們定位處理任務所在的線程池
		executor.setThreadNamePrefix("RaviAsync-");
		executor.initialize();
		return executor;
	}
}

4. Controller 測試代碼

@Slf4j
@RestController
public class TestController {
	@Autowired
	private RedisLookupService service;

	@GetMapping("/en")
	public String enqueueTry() throws InterruptedException {
		long start = System.currentTimeMillis();
		CustomerVo c1 = new CustomerVo(1, "A031", "馬哲", "男");
		CustomerVo c2 = new CustomerVo(2, "A039", "馬王", "男");
		CustomerVo c3 = new CustomerVo(3, "A040", "馬麗", "女");
		CompletableFuture<Long> future1 = service.enqueueCustomer(c1);
		CompletableFuture<Long> future2 = service.enqueueCustomer(c3);
		CompletableFuture<Long> future3 = service.enqueueCustomer(c2);
		CompletableFuture.allOf(future1, future2, future3).join();
		long end = System.currentTimeMillis();
		log.info("complete test: {}s",(float)(end - start) / 1000);
		return "ok";
	}

	@GetMapping("/qn")
	public String dequeueTry() {
		long start = System.currentTimeMillis();
		CompletableFuture<CustomerVo> customer1 = service.dequeueCustomer();
		CompletableFuture<CustomerVo> customer2 = service.dequeueCustomer();
		CompletableFuture<CustomerVo> customer3 = service.dequeueCustomer();
		CompletableFuture.allOf(customer1, customer2, customer3).join();
		long end = System.currentTimeMillis();
		log.info("complete test: {}s",(float)(end - start) / 1000);
		return "ok";
	}
}

/en的測試結果：
圖1

圖2

/qn的測試結果：

由此可以發現，多線程已經啟動。

5. 日志設置

yml配置

logging:
  config: classpath:logback.xml
  level:
    com.ravi.mapper: trace

xml配置

<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
    <!--定義日志文件的存儲地址 勿在 LogBack 的配置中使用相對路徑-->
    <property name="LOG_HOME" value="/Users/ravi/codes/myproject/log"/>
    <!-- 定義日志格式  -->
    <property name="LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%-5level] [%thread] [%-30.30logger{30}] %msg%n"/>
    <!-- 控制台輸出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化輸出：%d表示日期，%thread表示線程名，%-5level：級別從左顯示5個字符寬度%msg：日志消息，%n是換行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
    </appender>
    <!-- 按照每天生成日志文件 -->
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件輸出的文件名-->
            <FileNamePattern>${LOG_HOME}/Slf4j_%d{yyyy-MM-dd}.log</FileNamePattern>
            <!--日志文件保留天數-->
            <MaxHistory>30</MaxHistory>
        </rollingPolicy>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化輸出：%d表示日期，%thread表示線程名，%-5level：級別從左顯示5個字符寬度%msg：日志消息，%n是換行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
        <!--日志文件最大的大小-->
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <MaxFileSize>10MB</MaxFileSize>
        </triggeringPolicy>
    </appender>

    <!-- 日志輸出級別 -->
    <logger name="org.springframework" level="INFO"/>
    <logger name="com.ravi.mapper" level="INFO"/>
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="FILE"/>
    </root>
</configuration>

參考文章大數據從業者