当前位置：首页 > news >正文

Redis+定式任务实现简易版消息队列

news 来源：原创 2024/7/7 20:02:22

Redis是一个开源的内存中数据结构存储系统，通常被用作数据库、缓存和消息中间件。
Redis主要将数据存储在内存中，因此读写速度非常快。
支持不同的持久化方式，可以将内存中的数据定期写入磁盘，保证数据持久性。
redis本身就有自己的发布与订阅功能，实现简单的消息队列系统。

这里是另一种方式实现消息队列的机制，使用定式任务。
代码实现：
1、启动类开启定是任务

@SpringBootApplication
@EnableScheduling //启动类开启定时任务
public class AccountApplication {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(AccountApplication.class, args);}
}

2、准备redis缓存工具类

/*** Redis的配置类*/
@Configuration
public class RedisConfiguration {private static final Logger logger = LoggerFactory.getLogger(RedisConfiguration.class);public RedisConfiguration() {logger.info("创建缓存配置类：RedisConfiguration");}@Beanpublic RedisTemplate<String, Serializable> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(redisConnectionFactory);redisTemplate.setKeySerializer(RedisSerializer.string());redisTemplate.setValueSerializer(RedisSerializer.json());return redisTemplate;}
}/*** Redis的缓存接口*/
public interface CacheInterface {//往队列中存入数据void saveDate(Integer i, LinkedHashMap<String, Object> map);//从队列中取数据LinkedHashMap<String, Object> getDate(Integer i);//清除队列中的数据void deleteDate(Integer i);//检查队列中剩余的数据Set<String> checkDataKey();
}/*** Redis的缓存接口的实现类*/
@Repository
public class CacheRepositoryImpl implements CacheInterface {private static final Logger logger = LoggerFactory.getLogger(CacheRepositoryImpl.class);@Autowiredprivate RedisTemplate<String, Serializable> redisTemplate;@Overridepublic void saveDate(Integer i, LinkedHashMap<String, Object> map) {String key = YOU_KEY + i;ValueOperations<String, Serializable> opsForValue = redisTemplate.opsForValue();opsForValue.set(key, map);}@Overridepublic LinkedHashMap<String, Object> getDate(Integer i) {String key = YOU_KEY + i;ValueOperations<String, Serializable> opsForValue = redisTemplate.opsForValue();Serializable serializable = opsForValue.get(key);if(serializable == null ){return null;}return (LinkedHashMap<String, Object>) serializable;}@Overridepublic void deleteDate(Integer i) {String key = YOU_KEY + i;redisTemplate.delete(key);}@Overridepublic Set<String> checkDataKey() {Set<String> keys = redisTemplate.keys(YOU_KEY  + "*");return keys;}
}

3、准备指针工具类与存取方法

    
public class TaskCount {public static Integer save = 0;public static Integer get = 0;
}@Service
public class ServiceImpl implements Service {@Autowiredprivate CacheInterface cacheImpl;//存入队列@Overridepublic JsonResult saveDate(LinkedHashMap<String, Object> map ) {cacheImpl.saveDate(TaskCount.save,map);TaskCount.save++;return JsonResult.ok();}//处理数据@Overridepublic JsonResult handleDate(LinkedHashMap<String, Object> map ) {//处理逻辑}
}

4、定式任务工具类充当消息的发布

@Component
public class TaskUtil implements CommandLineRunner {@Autowiredprivate CacheInterface cacheImpl;@Autowiredprivate Service serviceImpl;//定时任务处理,每5000毫秒@Scheduled(fixedRate = 5000)public void handleData(){LinkedHashMap<String, Object> map = cacheImpl.getDate(TaskCount.get);if(map != null){try {serviceImpl.handleData(map);cacheImpl.deleteDate(TaskCount.get);TaskCount.get++;}catch (Exception e){System.out.println(e.getMessage());}}}//每小时启动一次@Scheduled(fixedRate = 3600000)public void handleData(){handleCount(cacheImpl);}//开机加载启动@Overridepublic void run(String... args) {handleCount(cacheImpl);}}//归置双指针public static void handleCount(CacheInterface cacheImpl){try {Set<String> keys = cacheImpl.checkDataKey();int max = 0;int min = Integer.MAX_VALUE;for (String key : keys) {String[] split = key.split(":");int a = Integer.parseInt(split[split.length-1]);if(a>max){max = a;}if(a<min){min = a;}}if(max > 0 ){max = max+1;}else {min = 0;}TaskCount.get = min;TaskCount.save = max;}catch (Exception e){System.out.println(e.getMessage());}}
}