首页 > 数据库 >spring boot + Redis实现消息队列-生产消费者

spring boot + Redis实现消息队列-生产消费者

时间:2022-12-14 18:36:24浏览次数:59  
标签:return spring boot Redis queue configuration consumer public redisTemplate


实现思路:

Redis本身提供了一个发布/订阅模式,但生产消费者模式需要我们自己去实现。

  1. 利用Redis中的队列,将新消息放入名称为xx的队列末尾,完成消息生产者。
  2. 启动一个线程,使用​​brpop​​命令循环从xx队列取第一个元素,获得消息,调用注册的消费者执行业务逻辑。

redis 集成:

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

配置RedisTemplate

/**
* redis配置
*
*/
@Configuration
public class RedisConfig {


/**
* retemplate相关配置
* @param factory
* @return
*/
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {

RedisTemplate<String, Object> template = new RedisTemplate<>();
// 配置连接工厂
template.setConnectionFactory(factory);

//使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值(默认使用JDK的序列化方式)
Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);

ObjectMapper om = new ObjectMapper();
// 指定要序列化的域,field,get和set,以及修饰符范围,ANY是都有包括private和public
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 指定序列化输入的类型,类必须是非final修饰的,final修饰的类,比如String,Integer等会跑出异常
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jacksonSeial.setObjectMapper(om);

// 值采用json序列化
template.setValueSerializer(jacksonSeial);
//使用StringRedisSerializer来序列化和反序列化redis的key值
template.setKeySerializer(new StringRedisSerializer());

// 设置hash key 和value序列化模式
template.setHashKeySerializer(new StringRedisSerializer());
template.setHashValueSerializer(jacksonSeial);
template.afterPropertiesSet();

return template;
}

/**
* 对hash类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForHash();
}

/**
* 对redis字符串类型数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForValue();
}

/**
* 对链表类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForList();
}

/**
* 对无序集合类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForSet();
}

/**
* 对有序集合类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForZSet();
}

/**
* redis消息监听器容器
* 可以添加多个监听不同话题的redis监听器,只需要把消息监听器和相应的消息订阅处理器绑定,该消息监听器
* 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
* @param connectionFactory
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter webMessage) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅了一个叫chat的通道
container.addMessageListener(webMessage, new PatternTopic("webMessage"));

return container;
}

@Bean
MessageListenerAdapter webMessage(MessageReceiver messageReceiver) {

//给messageListenerAdapter 传入一个消息接受的处理器,利用反射的方法调用“receiveMessage”
//不填defaultListenerMethod默认调用handleMessage
return new MessageListenerAdapter(messageReceiver, "receiveMessage");
}

/**
* 配置redis消息队列消费者容器
*
* @param redisTemplate redis
* @return 消费者容器
*/
@Autowired
NormalMessageReceiver normalMessageReceiver;


@Bean(initMethod = "init", destroyMethod = "destroy")
public RedisMqConsumerContainer redisMqConsumerContainer(RedisTemplate<String, Object> redisTemplate) {
RedisMqConsumerContainer config = new RedisMqConsumerContainer(redisTemplate);
config.addConsumer(QueueConfiguration.builder()
.queue("normalMessage")
.consumer(normalMessageReceiver)
.build());

return config;
}

消息-生产者

public class QueueSender {
private RedisTemplate<Object, Object> redisTemplate;

public QueueSender(RedisTemplate<Object, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}

public void sendMsg(String queue, Object msg) {
redisTemplate.opsForList().leftPush(queue, msg);

}
}

消息-消费者

 消费接口类

public interface MsgConsumer {
void onMessage(Object message);

void one rror(Object msg, Exception e);
}

消息监听类

class QueueListener implements Runnable {
public static final Logger log = LoggerFactory.getLogger(QueueListener.class);
private RedisTemplate<Object, Object> redisTemplate;
private String queue;
private MsgConsumer consumer;

public QueueListener(RedisTemplate<Object, Object> redisTemplate, String queue, MsgConsumer consumer) {
this.redisTemplate = redisTemplate;
this.queue = queue;
this.consumer = consumer;
}

@Override
public void run() {
log.info("QueueListener start...queue:{}", queue);
while (RedisMqConsumerContainer.run) {
try {
Object msg = redisTemplate.opsForList().rightPop(queue, 30, TimeUnit.SECONDS);
if (msg != null) {
try {
consumer.onMessage(msg);
} catch (Exception e) {
consumer.onError(msg, e);
}
}
} catch (QueryTimeoutException ignored) {
} catch (Exception e) {
if (RedisMqConsumerContainer.run) {
log.error("Queue:{}", queue, e);
} else {
log.info("QueueListener exits...queue:{}", queue);
}
}
}
}
}

消息-消费者容器

配置类

public class QueueConfiguration {
/**
* 队列名称
*/
private String queue;
/**
* 消费者
*/
private MsgConsumer consumer;

private QueueConfiguration() {
}

public static Builder builder() {
return new Builder();
}

String getQueue() {
return queue;
}

MsgConsumer getConsumer() {
return consumer;
}

public static class Builder {
private QueueConfiguration configuration = new QueueConfiguration();

public QueueConfiguration defaultConfiguration(MsgConsumer consumer) {
configuration.consumer = consumer;
configuration.queue = consumer.getClass().getSimpleName();
return configuration;
}

public Builder queue(String queue) {
configuration.queue = queue;
return this;
}

public Builder consumer(MsgConsumer consumer) {
configuration.consumer = consumer;
return this;
}

public QueueConfiguration build() {
if (configuration.queue == null || configuration.queue.length() == 0) {
if (configuration.consumer != null) {
configuration.queue = configuration.getClass().getSimpleName();
}
}
return configuration;
}

}
}

消息--消费者容器

public class RedisMqConsumerContainer {
private static final Logger log = LoggerFactory.getLogger(RedisMqConsumerContainer.class);
private Map<String, QueueConfiguration> consumerMap = new HashMap<>();
private RedisTemplate<Object, Object> redisTemplate;
static boolean run;
private ExecutorService exec;

public RedisMqConsumerContainer(RedisTemplate<Object, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}

public void addConsumer(QueueConfiguration configuration) {
if (consumerMap.containsKey(configuration.getQueue())) {
log.warn("Key:{} this key already exists, and it will be replaced", configuration.getQueue());
}
if (configuration.getConsumer() == null) {
log.warn("Key:{} consumer cannot be null, this configuration will be skipped", configuration.getQueue());
}
consumerMap.put(configuration.getQueue(), configuration);
}

public void destroy() {
run = false;
this.exec.shutdown();
log.info("QueueListener exiting.");
while (!this.exec.isTerminated()) {

}
log.info("QueueListener exited.");
}

public void init() {
run = true;
this.exec = Executors.newCachedThreadPool(r -> {
final AtomicInteger threadNumber = new AtomicInteger(1);
return new Thread(r, "RedisMQListener-" + threadNumber.getAndIncrement());
});
consumerMap = Collections.unmodifiableMap(consumerMap);
consumerMap.forEach((k, v) -> exec.submit(new QueueListener(redisTemplate, v.getQueue(), v.getConsumer())));
}

}

配置消费者

public class NormalMessageReceiver implements MsgConsumer {
private static Logger log = LoggerFactory.getLogger(TestListener.class);

@Override
public void onMessage(Object message) {
log.info("收到消息:" + message);

}

@Override
public void one rror(Object msg, Exception e) {
log.error("发生错误,消息:{}", msg, e);
}
}

测试代码如下:

//开启定时器功能
@EnableScheduling
@Component
public class MessageSender {

@Autowired
private QueueSender queueSender;

@Scheduled(fixedDelay = 5000)
public void sendMessageXXX() {
//stringRedisTemplate.convertAndSend("chat1", String.valueOf(Math.random()));
//producer.publishMessageXXX("队列消息1xxxxx"+System.currentTimeMillis());

for (int i = 0; i < 20; i++) {
queueSender.sendMsg("TEST0","hello quit~~~~,序号:"+i);
queueSender.sendMsg("TEST1","hello quit~~~~,序号:"+i);
queueSender.sendMsg("TEST2","hello quit~~~~,序号:"+i);
}

}
}

上述代码,有些来源与网络,在下只是在这里做个记忆。我进行删除操作。

redis 作为消息中间件只是在小系统中可以稍微用一下,如果对系统要求高的请使用真正的消息中间件。RabbitMQ和RocketMQ

标签:return,spring,boot,Redis,queue,configuration,consumer,public,redisTemplate
From: https://blog.51cto.com/u_15461374/5938161

相关文章

  • redis5-cluster 集群搭建
    1、安装环境信息centos7redis52、整体集群信息#以直接在一台机器上实现上述的伪集群,因为端口号特意设置为不同的。#重点:不论机器多少,对于部署过程都是一样的,只不过是在不......
  • gateway动态路由实现 mysql+redis 实现
    前言大家都知道咱们在通常是使用配置文件来实现配置,但是这样就有一个弊端,就是每次修改的时候都要去重启来实现,并且管理起来非常麻烦,所有就有了这种实现方式。现在的实现方式......
  • SpringBoot 设置动态定时任务
    前言SpringBoot项目中简单使用定时任务,不过由于要借助cron表达式且都提前定义好放在配置文件里,不能在项目运行中动态修改任务执行时间,不是太灵活,改文章是主要是实现在固定的......
  • spring boot 实现Mysql数据脚本导出和数据库脚本的导入
    前言在开发过程中这样一个需求,有些数据需要从数据库导出,然后导入到另外的数据库中。数据导出@SneakyThrowspublicStringexport(){//获取数据库连接对象......
  • 【Java】Spring Cache 缓存
    SpringCache一、Spring缓存抽象Spring从3.1开始定义了org.springframework.cache.Cache和org.springframework.cache.CacheManager接口来统一不同的缓存技术;并支......
  • Spring Integration的网络通量支持
    WebFluxSpring集成模块()允许以反应方式执行HTTP请求和处理入站HTTP请求。​​spring-integration-webflux​​您需要将此依赖项包含在项目中:<dependency><groupI......
  • 一文搞懂 Redis 架构演化之路
    作者:ryetan,腾讯CSIG后台开发工程师现如今Redis变得越来越流行,几乎在很多项目中都要被用到,不知道你在使用Redis时,有没有思考过,Redis到底是如何稳定、高性能地提供服务......
  • SpringMVC学习
    SpringMVC学习1.回顾MVC1.1什么是MVCMVC是模型(Model)、视图(View)、控制器(Controller)的简写,是一种软件设计规范。是将业务逻辑、数据、显示分离的方法来组织代码......
  • spring mvc环境之异常错误码的统一返回(十五)
    1.根据不同的请求方式,返回页面或json数据1).创建统一权限异常处理类,所有的权限异常走一个端口2).根据请求方式不同返回不同数据,页面请求返回403未授权页面,ajax请......
  • spring学习-1.使用Maven构建spring项目
    1.新建一个Maven项目​​​​​​项目的结构图​​​​2.配置pom.xml,引入项目需要的依赖,这里引入了spring,junit<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns......