---

1.简介

使用pub/pub/sub主题订阅者模式，可以实现1:N的消息队列。

缺点：在消费者下线的情况下，生产的消息会丢失。此场景，建议用MQ。

2.代码

被请求的接口（被订阅者），使用redis队列,放于队列中。

package com.airboot.bootdemo.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

@RestController

@RequestMapping(“/SentRedisController”)

public class SentRedisController {

@Autowired

private RedisTemplate redisTemplate;

/**

• redis生产者测试

• @param data

• @return

*/

@GetMapping(“/send1”)

String send1(String data) {

redisTemplate.convertAndSend(“testkafka”, data);

return “success”;

}

/**

• redis生产者测试

• @param data

• @return

*/

@GetMapping(“/send2”)

String send2(String data) {

redisTemplate.convertAndSend(“testkafka1”, data);

return “success”;

}

}

配置监听器。监听队列。

package com.airboot.bootdemo.config;

import com.airboot.bootdemo.controller.RedisSubscriber;

import com.airboot.bootdemo.controller.RedisSubscriberTwo;

import org.springframework.beans.factory.annotation.Autowired;

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.listener.PatternTopic;

import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;

import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration

public class RedisConfig {

@Autowired

private RedisTemplate redisTemplate;

//序列化通用设置

@Bean

public RedisTemplate redisTemplateInit() {

Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

// 设置序列化Key的实例化对象

redisTemplate.setKeySerializer(new StringRedisSerializer());

// 设置序列化Value的实例化对象

redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

redisTemplate.setHashKeySerializer(jackson2JsonRedisSerializer);

redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);

return redisTemplate;

}

//配置监听 配置使类RedisSubscriber RedisSubscriberTwo 去监听testkafka1，

//testkafka这两个队列

@Bean

RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,

RedisSubscriber listenerAdapter,

RedisSubscriberTwo listenerAdapter2){

RedisMessageListenerContainer container = new RedisMessageListenerContainer();

container.setConnectionFactory(connectionFactory);

//订阅了一个叫chat 的通道

container.addMessageListener(listenerAdapter, new PatternTopic(“testkafka”));

container.addMessageListener(listenerAdapter, new PatternTopic(“testkafka1”));//配置要订阅的订阅项

container.addMessageListener(listenerAdapter2, new PatternTopic(“testkafka”));//配置要订阅的订阅项

//这个container 可以添加多个 messageListener

return container;

}

}

可以多个消费者通过上面配置多个类的监听就可以。

package com.airboot.bootdemo.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.connection.Message;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import org.springframework.stereotype.Component;

@Component

public class RedisSubscriberTwo extends MessageListenerAdapter {

@Autowired

private RedisTemplate<String, String> redisTemplate;

@Override

public void onMessage(Message message, byte[] bytes) {

System.out.println(message);

byte[] body = message.getBody();

byte[] channel = message.getChannel();

String msg = redisTemplate.getStringSerializer().deserialize(body);

String topic = redisTemplate.getStringSerializer().deserialize(channel);

System.out.println(“监听到topic为2” + topic + “的消息：” + msg);

}

}

package com.airboot.bootdemo.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.connection.Message;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import org.springframework.stereotype.Component;

@Component

public class RedisSubscriber extends MessageListenerAdapter {

//

@Autowired

private RedisTemplate<String, String> redisTemplate;

@Override

public void onMessage(Message message, byte[] bytes) {

System.out.println(message);

byte[] body = message.getBody();

byte[] channel = message.getChannel();

String msg = redisTemplate.getStringSerializer().deserialize(body);

String topic = redisTemplate.getStringSerializer().deserialize(channel);

System.out.println(“监听到topic为” + topic + “的消息：” + msg);

}

}

三.延时队列

1.简介

上面的例子我们已经了一个简易的消息队列。我们继续思考一个现实的场景，假定这些是一些游戏商品，它需要添加"延迟销售"特性，在未来某个时候才可以开始处理这些游戏商品数据。 那么要实现这个延迟的特性，我们需要修改现有队列的实现。

1. 在消息数据的信息中包含延迟处理消息的执行时间，如果工作进程发现消息的执行时间还没到，那么它将会在短暂的等待之后重新把消息数据推入队列中。(延迟发送消息)

2. 使用有序集合来存储这些需要延时消费的消息数据，将任务的执行时间设置为分值，在开启一个工作进程查找有序集合里面是否有可以立刻执行的任务，如果有的话就从有序集合中移除消息并且消费。

先自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数初中级Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则近万的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《Java开发全套学习资料》送给大家，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频

如果你觉得这些内容对你有帮助，可以扫码领取！

最后

分享一些资料给大家，我觉得这些都是很有用的东西，大家也可以跟着来学习，查漏补缺。

《Java高级面试》

《Java高级架构知识》

《算法知识》

觉得这些内容对你有帮助，可以扫码领取！**

最后

分享一些资料给大家，我觉得这些都是很有用的东西，大家也可以跟着来学习，查漏补缺。

《Java高级面试》

[外链图片转存中…(img-qtgZahB6-1711441463371)]

《Java高级架构知识》

[外链图片转存中…(img-PQ9LWJbe-1711441463372)]

《算法知识》

[外链图片转存中…(img-DvabLGGc-1711441463372)]

需要更多Java资料的小伙伴可以帮忙点赞+关注，点击传送门，即可免费领取！