Spring boot 利用Redis實作隊列消息監聽簡單示例

我的應用場景：當某個業務系統接收到MQ消息需要按照順序進行執行時，且收到的消息間隔時間過短時，可以把需要執行的消息放到隊列裡面進行逐個消費，因為對消息執行的代碼加鎖是不行的，因為消息間隔時間小，容易出錯，隻能對消息再進行一層封裝，然後執行.,這種情況隻适用于消息不是必須實時消費。

1.Redis配置

主要配置序列化和反序列化使用的方式.

package com.example.demo.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
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;

/**
 * 內建Redis
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();

        redisTemplate.setConnectionFactory(factory);
        redisTemplate.afterPropertiesSet();
        setSerializer(redisTemplate);
        return redisTemplate;
    }

    private void setSerializer(RedisTemplate<String, String> template) {
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
    }
}

           

2.按照存儲Redis資料

這裡面的value實際就是得到的消費消息.

@SpringBootTest
@RunWith(SpringRunner.class)
class DemoApplicationTests {
	private static final String key = "test007";

	@Autowired
	private RedisTemplate redisTemplate;

	@Test
	void contextLoads() {
		System.out.println(redisTemplate == null);
		redisTemplate.opsForList().leftPush(key , 12345);
		redisTemplate.opsForList().leftPush(key , 125678);
		Long size = redisTemplate.opsForList().size(key);
		System.out.println(size);

/*		Object object = redisTemplate.opsForList().rightPop(key);
		Object pop = redisTemplate.opsForList().rightPop(key);
		System.out.println(object);
		System.out.println(pop);
		Long temp = redisTemplate.opsForList().size(key);
		System.out.println(temp);*/

	}

}
           

監聽器進行監聽消費

定時器執行間隔，可以依照不同業務場景進行配置,用key的size進行判斷redis是否存儲了消費消息。因為是單線程，是以不會存線上程安全問題.

定時器執行需要在啟動類加上@EnableScheduling注解.

@Component
@Slf4j
public class MyTask001 {
	
	//這裡需要把key提取成常量， 因為存取需要同一個key
    private static final String key = "test007";
	
	//辨別符，判斷redis裡面是否有消息.
    private static  Long keyNum = 0L;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 定時執行隊列資訊消費
     */
    @Scheduled(cron = "55 * * * * ?")
    public void testTask(){
        try {
            keyNum = redisTemplate.opsForList().size(key);
            while(keyNum > 0){
                Object object = redisTemplate.opsForList().rightPop(key);
                //執行的業務代碼,單線程，是以不必對業務代碼進行加鎖.且是順序執行.
                System.out.println(object);
                //再次擷取辨別符
                keyNum = redisTemplate.opsForList().size(key);
            }
        } catch (Exception e) {
            log.info("定時器異常資訊,{}",e);
        }
    }

}
           

附錄: