1.引入依赖
<!-- 整合redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<version>3.0.0</version>
</dependency>
2.添加redis配置
spring:
#缓存服务
redis:
client-name: redis
host: ip
port: 6379
password: you password
timeout: 3000
jedis:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
3.redistemplate配置
package com.hhmt.delivery.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;
/**
* 辉煌明天
* FileName: RedisConfiguration
* Author: huachun
* email: huachun_w@163.com
* Date: 2021/11/5 15:57
* Description: Redis配置类
*/
@Configuration
public class RedisConfiguration {
@Bean
@SuppressWarnings("all")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
// 配置连接工厂
template.setConnectionFactory(factory);
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);
template.afterPropertiesSet();
return template;
}
}
4.消息发布
使用 StringRedisTemplate或者RedisTemplate的convertAndSend(channel, message)方法即可,
其中channel代表消息信道也可以理解为主题,message表示发布的内容。
消息发布类 ,模拟service层的逻辑处理
package com.hhmt.delivery.aop;
import com.alibaba.fastjson.JSON;
import com.hhmt.delivery.model.ServiceMessageInfo;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
/**
* @author huachun
* @version 1.0
* @description: 拦截持久化操作推送消息切面
* @email huachun_w@163.com
* @date 2023-02-03 14:18
*/
@Aspect
@Component
@Slf4j
public class ServiceMessageAop {
@Autowired
private RedisTemplate redisTemplate;
public void doAfter(String channel, String message) {
redisTemplate.convertAndSend("mq_01", message);
redisTemplate.convertAndSend("mq_02", "hello");
redisTemplate.convertAndSend("mq_03", "mq");
}
}
5.消息接受,处理业务
消息监听注册配置,把消息监听注册到容器里面
package com.hhmt.delivery.config;
import com.hhmt.delivery.mq.MessageReceiver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
/**
* @author huachun
* @version 1.0
* @description: TODO
* @email huachun_w@163.com
* @date 2023-02-03 17:53
*/
@Configuration
public class RedisConsumeConfig {
/**
* 注入消息监听容器
*
* @param connectionFactory 连接工厂
* @param listenerAdapter 监听处理器1
* @param listenerAdapter 监听处理器2 (参数名称需和监听处理器的方法名称一致,因为@Bean注解默认注入的id就是方法名称)
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter,
MessageListenerAdapter listenerAdapter2) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅一个叫mq_01 的信道
container.addMessageListener(listenerAdapter, new PatternTopic("mq_01"));
//订阅一个叫mq_02 的信道
container.addMessageListener(listenerAdapter2, new PatternTopic("mq_02"));
//这个container 可以添加多个 messageListener
return container;
}
/**
* 消息监听处理器1
*
* @param messageReceiver 处理器类
* @return
*/
@Bean
MessageListenerAdapter listenerAdapter(MessageReceiver messageReceiver) {
//给messageListenerAdapter 传入一个消息接收的处理器,利用反射的方法调用“receiveMessage”
return new MessageListenerAdapter(messageReceiver, "receiveMessage"); //receiveMessage:接收消息的方法名称
}
/**
* 消息监听处理器2
*
* @param receiver 处理器类
* @return
*/
@Bean
MessageListenerAdapter listenerAdapter2(MessageReceiver receiver) {
//给messageListenerAdapter 传入一个消息接收的处理器,利用反射的方法调用“receiveMessage2”
return new MessageListenerAdapter(receiver, "receiveMessage2"); //receiveMessage:接收消息的方法名称
}
}
6.自定义消息处理器
package com.hhmt.delivery.mq;
import org.springframework.stereotype.Component;
/**
* @author huachun
* @version 1.0
* @description: TODO
* @email huachun_w@163.com
* @date 2023-02-03 18:08
*/
@Component
public class MessageReceiver {
/**
* 接收消息的方法1
**/
public void receiveMessage(String message) {
System.out.println("receiveMessage接收到的消息:" + message);
}
/**
* 接收消息的方法2
**/
public void receiveMessage2(String message) {
System.out.println("receiveMessage2接收到的消息:" + message);
}
}
第二种方式:
通过上面方式虽然可以发布和订阅消息,但是消息的内容局限与String类型,有时候我们需要发布自定义类型的数据,需要用到下面这种方式
1.更改容器配置
package com.hhmt.delivery.config;
import com.hhmt.delivery.mq.RedisConsumerSubscribe;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
/**
* 辉煌明天
* FileName: RedisConfig
* Author: huachun
* email: huachun_w@163.com
* Date: 2021/11/15 17:19
* Description: redis配置类
*/
@Configuration
public class RedisConfig {
@Value("${spring.profiles.active}")
private String env;
@Autowired
private RedisConsumerSubscribe redisConsumerSubscribe;
/**
* @Description: redis消息监听配置
* @Author: huachun
* @Date: 2021/11/15 17:20
* @return: org.springframework.data.redis.listener.RedisMessageListenerContainer
**/
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅频道,通配符*表示任意多个占位符
container.addMessageListener(redisConsumerSubscribe, new PatternTopic("ocpx"));
return container;
}
@Bean
public MessageListenerAdapter listenerAdapter() {
return new MessageListenerAdapter(redisConsumerSubscribe);
}
}
2.更改redis配置
@Bean
@SuppressWarnings("all")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
template.setConnectionFactory(factory);
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);
jackson2JsonRedisSerializer.setObjectMapper(ObjectMapperConfig.objectMapper);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
template.setKeySerializer(stringRedisSerializer);
template.setHashKeySerializer(stringRedisSerializer);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
3.消息接受与数据转换
package com.hhmt.delivery.mq;
import com.hhmt.delivery.config.ObjectMapperConfig;
import com.hhmt.delivery.model.ServiceMessage;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
/**
* @author huachun
* @version 1.0
* @description: TODO
* @email huachun_w@163.com
* @date 2023-04-21 14:32
*/
public class RedisConsumerSubscribe implements MessageListener {
@Override
public void onMessage(Message message, byte[] bytes) {
Jackson2JsonRedisSerializer<ServiceMessage> jacksonSerializer = new Jackson2JsonRedisSerializer<>(ServiceMessage.class);
jacksonSerializer.setObjectMapper(ObjectMapperConfig.objectMapper);
ServiceMessage message1 = jacksonSerializer.deserialize(message.getBody());
System.out.println("ServiceMessage:" + message1);
System.out.println("订阅频道:" + new String(message.getChannel()));
System.out.println("接收数据:" + new String(message.getBody()));
}
}
原文参考:springboot:整合redis之消息队列_springboot redis 队列_yololee_的博客-CSDN博客
原文参考:Redis 消息订阅(MessageListener接口) - Spring Data Redis 教程