SpringBoot中使⽤Redis的发布订阅模式redis的发布订阅模式，使发布者和订阅者完全解耦
l and application.properties
<!-- 引⼊redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<exclusions>
<exclusion>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
spring:
redis:
#数据库索引
database: 5
host: 127.0.0.1
port: 6379
password: 123456
jedis:
pool:
#最⼤连接数
max-active: 8
#最⼤阻塞等待时间(负数表⽰没限制)
#最⼤空闲
max-idle: 8
#最⼩空闲
min-idle: 0
2.消息发布者、消息处理者POJO、redis消息容器以及redis注⼊IOC容器@Configuration //相当于xml中的beans
public class RedisConfig {
/**
* redis消息容器
* 可以添加多个监听不同话题的redis，只需要把消息和相应的消息订阅处理器绑定，该消息
* 通过反射技术调⽤消息订阅处理器的相关⽅法进⾏⼀些业务处理
* @param connectionFactory
* @param listenerAdapter
* @return
*/
@Bean //相当于xml中的bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter,
MessageListenerAdapter listenerAdapter2) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅了⼀个叫chat 的通道
container.addMessageListener(listenerAdapter, new PatternTopic("chat"));
container.addMessageListener(listenerAdapter2, new PatternTopic("chat2"));
//这个container 可以添加多个 messageListener
return container;
}
/**
* 消息适配器，绑定消息处理器，利⽤反射技术调⽤消息处理器的业务⽅法
* @param receiver
* @return
*/
@Bean
MessageListenerAdapter listenerAdapter(MessageReceiver receiver) {
//这个地⽅是给messageListenerAdapter 传⼊⼀个消息接受的处理器，利⽤反射的⽅法调⽤“receiveMessage”
//也有好⼏个重载⽅法，这边默认调⽤处理器的⽅法叫handleMessage 可以⾃⼰到源码⾥⾯看
return new MessageListenerAdapter(receiver, "receiveMessage");
}
@Bean
MessageListenerAdapter listenerAdapter2(MessageReceiver receiver) {
//这个地⽅是给messageListenerAdapter 传⼊⼀个消息接受的处理器，利⽤反射的⽅法调⽤“receiveMessage2”
//也有好⼏个重载⽅法，这边默认调⽤处理器的⽅法叫handleMessage 可以⾃⼰到源码⾥⾯看
return new MessageListenerAdapter(receiver, "receiveMessage2");
}
/**redis 读取内容的template */
@Bean
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
}
@Bean
RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory) {
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class); ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
jackson2JsonRedisSerializer.setObjectMapper(om);
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
template.setConnectionFactory(connectionFactory);
template.setKeySerializer(jackson2JsonRedisSerializer);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashKeySerializer(jackson2JsonRedisSerializer);
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
MessageListenerAdapter通过反射使普通的POJO就可以处理消息。具体情况见MessageListenerAdapter的onMessage⽅法。
3.消息发布者
@EnableScheduling //开启定时器功能
@Component
public class MessageSender {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Scheduled(fixedRate = 2000) //间隔2s 通过StringRedisTemplate对象向redis消息队列chat频道发布消息
public void sendMessage(){
}
}
4.普通的消息处理器POJO
@Component
public class MessageReceiver {
/**接收消息的⽅法*/
public void receiveMessage(String message){
System.out.println("收到⼀条chat的消息："+message);
}
/**接收消息的⽅法*/
public void receiveMessage2(String message){
System.out.println("收到⼀条chat2的消息："+message);
}
}
MessageListenerAdapter通过反射调⽤receiveMessage⽅法处理消息
5.其他⽅式(参考)
配置
ample.disConfiBack;
/**
* redis消息队列配置-订阅者
*/
import t.annotation.Bean;
import t.annotation.Configuration;
import org.tion.RedisConnectionFactory;
import org.dis.listener.PatternTopic;
import org.dis.listener.RedisMessageListenerContainer;
import org.dis.listener.adapter.MessageListenerAdapter;
import urrent.CountDownLatch;
/**
* redis消息队列配置-订阅者
*/
@Configuration
public class RedisMessageListener {
/**
* 创建连接⼯⼚
* @param connectionFactory
* @param listenerAdapter
* @return
*/
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter,
MessageListenerAdapter listenerAdapterTest2){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//接受消息的key
container.addMessageListener(listenerAdapter,new PatternTopic("phone"));
container.addMessageListener(listenerAdapterTest2,new PatternTopic("phoneTest2"));
return container;
}
/**
* 绑定消息监听者和接收监听的⽅法
* @param receiver
* @return
*/
@Bean
public MessageListenerAdapter listenerAdapter(ReceiverRedisMessage receiver){
return new MessageListenerAdapter(receiver,"receiveMessage");
}
/**
* 绑定消息监听者和接收监听的⽅法
* @param receiver
* @return
*/
@Bean
public MessageListenerAdapter listenerAdapterTest2(ReceiverRedisMessage receiver){
return new MessageListenerAdapter(receiver,"receiveMessage2");
}
/**
* 注册订阅者
* @param latch
* @return
*/
@Bean
ReceiverRedisMessage receiver(CountDownLatch latch) {
return new ReceiverRedisMessage(latch);
}
/**
* 计数器，⽤来控制线程
* @return
*/
springframework和springboot@Bean
public CountDownLatch latch(){
return new CountDownLatch(1);//指定了计数的次数 1
}
}
消息处理
ample.disConfiBack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import urrent.CountDownLatch;
/**
* 注⼊消息接受类
*/
public class ReceiverRedisMessage {
private static final Logger log = Logger(ReceiverRedisMessage.class);
private CountDownLatch latch;
@Autowired
public ReceiverRedisMessage(CountDownLatch latch) {
this.latch = latch;
}
/**
* 队列消息接收⽅法
*
* @param jsonMsg
*/
public void receiveMessage(String jsonMsg) {
log.info("[开始消费REDIS消息队列phone数据...]");
try {
System.out.println(jsonMsg);
log.info("[消费REDIS消息队列phone数据成功.]");
} catch (Exception e) {
<("[消费REDIS消息队列phone数据失败，失败信息:{}]", e.getMessage());
}
}
/**
* 队列消息接收⽅法
*
* @param jsonMsg
*/
public void receiveMessage2(String jsonMsg) {
log.info("[开始消费REDIS消息队列phoneTest2数据...]");
try {
System.out.println(jsonMsg);
/
**
* 此处执⾏⾃⼰代码逻辑例如插⼊删除操作数据库等
*/
log.info("[消费REDIS消息队列phoneTest2数据成功.]");
} catch (Exception e) {
<("[消费REDIS消息队列phoneTest2数据失败，失败信息:{}]", e.getMessage()); }
}
}
测试
ample.disConfiBack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping
public class PublisherController {
private static final Logger log = Logger(PublisherController.class);
@Autowired
private RedisTemplate redisTemplate;
@GetMapping(value = "pub/{id}")
public String pubMsg(@PathVariable String id){
log.info("Publisher ");
return "success";
}
}
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论