Redis实现消息队列（2）——通过spring-boot-starter-data-redis发布和订阅消息

1.引入jar包

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> </dependency> <dependency> <groupId>io.github.xiaoyudeguang</groupId> <artifactId>easy-core</artifactId> <version>1.2.23-RELEASE</version> </dependency>

2.  RedisUtils工具类

@EasyBean(todo = { "redis工具类" }) public class RedisUtils { private static StringRedisTemplate stringRedisTemplate; @Autowired(required = false) public RedisUtils(StringRedisTemplate stringRedisTemplate) { RedisUtils.stringRedisTemplate = stringRedisTemplate; } /** * 获取redis连接工厂 * @return */ public static RedisConnectionFactory getConnectionFactory() { return stringRedisTemplate.getConnectionFactory(); } /** * 添加hash键值对 * @param key * @param value */ public static void putHashValue(String key, String value) { stringRedisTemplate.opsForValue().set(key, value); } /** * 添加hash键值对 * @param key * @param value */ public static void setHashValue(String key, String value) { stringRedisTemplate.opsForValue().set(key, value); } /** * 根据key得到hash值 * @param key * @param value */ public static String getHashValue(String key) { return stringRedisTemplate.opsForValue().get(key); } /** * list队列中左侧添加 * @param key * @param value * @return */ public static Long leftPush(String key, String value) { return stringRedisTemplate.opsForList().leftPush(key, value); } /** * list队列中右侧取出 * @param key * @return */ public static String rightPop(String key) { return stringRedisTemplate.opsForList().rightPop(key); } /** * 删除某个key * @param key * @return */ public static Boolean delete(String key) { return stringRedisTemplate.delete(key); } /** * 发布消息 * @param channel * @param message */ public static void publish(String channel, Object message){ stringRedisTemplate.convertAndSend(channel, JSON.toJSONString(message)); } /** * 发布消息 * @param channel * @param message * @return */ @SuppressWarnings("unchecked") public static <T> T deserialize(byte[] bytes){ return (T) stringRedisTemplate.getValueSerializer().deserialize(bytes); } }

3. 发布消息

RedisUtils.publish("自定义消息通道名称", "消息内容");

4.  注册订阅通道

@Configuration public class RedisConfig { @Autowired(required = false) private MessageListener listener; @Bean public RedisMessageListenerContainer redisMessageListenerContainer() { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(RedisUtils.getConnectionFactory()); container.addMessageListener(listener, new ChannelTopic("自定义消息通道名称1")); container.addMessageListener(listener, new ChannelTopic("自定义消息通道名称2")); return container; } }

5.  实现订阅接口

@EasyBean(todo = { "接收redis消息" }) public class RedisMsgListener implements MessageListener { @Override public void onMessage(Message message, byte[] pattern) { String key = RedisUtils.deserialize(pattern); //消息通道名称 String msg = RedisUtils.deserialize(message.getBody()); //消息内容 LogUtils.info(this, key + "订阅消息", message); } }