springboot（三）使用redis实现消息队列_redistemplate.convertandsend_一笑而过者也的博客

@Bean public StringRedisTemplate getRedisTemplate(RedisConnectionFactory redisConnectionFactory) { return new StringRedisTemplate(redisConnectionFactory);
消息生产者，注入redisTemplate,用convertAndSend发送消息
@Service
public class PublisherService {
    @Autowired
    private StringRedisTemplate redisTemplate;
    public String sendMessage(String name) {
        try {
            redisTemplate.convertAndSend("TOPIC_USERNAME", name);
            return "消息发送成功了";
        } catch (Exception e) {
            e.printStackTrace();
            return "消息发送失败了";
在controller中注入service,请求时发送消息 
@RestController
@RequestMapping("publisher")
public class PublisherController {
    @Autowired
    private PublisherService publisherService;
    @RequestMapping("{name}")
    public String sendMessage(@PathVariable("name") String name) {
        return publisherService.sendMessage(name);
消费者：创建一个接收消息的类，继承MessageListener，也可以不继承 
@Component
public class Receiver implements MessageListener {
    private static Logger logger = LoggerFactory.getLogger(Receiver.class);
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Override
    public void onMessage(Message message, byte[] pattern) {
        RedisSerializer<String> valueSerializer = redisTemplate.getStringSerializer();
        String deserialize = valueSerializer.deserialize(message.getBody());
        logger.info("收到的mq消息" + deserialize);
@Component
public class Receiver {
    private static Logger logger = LoggerFactory.getLogger(Receiver.class);
    public void receiveMessage(String message) {
        logger.info("收到的mq消息" + message);
消息订阅者配置类： 
@Configuration
@AutoConfigureAfter({Receiver.class})
public class SubscriberConfig {
     * 消息监听适配器，注入接受消息方法，输入方法名字 反射方法
     * @param receiver
     * @return
    @Bean
    public MessageListenerAdapter getMessageListenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage"); //当没有继承MessageListener时需要写方法名字
     * 创建消息监听容器
     * @param redisConnectionFactory
     * @param messageListenerAdapter
     * @return
    @Bean
    public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
        redisMessageListenerContainer.addMessageListener(messageListenerAdapter, new PatternTopic("TOPIC_USERNAME"));
        return redisMessageListenerContainer;
消费者和生产者配置已经完成了，启动springboot程序，用postman请求controller方法就可以了。 
实时内容请关注微信公众号，公众号与博客同时更新 
                    redis消息队列适合轻量级高并发的情况，比如秒杀，及时数据分析等。首先springboot配置文件配置如下：spring:  redis:    database: 1    host: 192.168.94.151    port: 6379    password: xuhaixing    jedis:      pool:            max-idle...
这篇记录一下redis的发布订阅模式的简单使用，各位看到此博客的小伙伴，如有不对的地方请及时通过私信我或者评论此博客的方式指出，以免误人子弟。多谢！
一、了解下发布订阅模式
借用菜鸟教程上的描述，Redis
				在介绍了 Redis 流的基本功能之后， 现在是时候使用这些功能来构建一些实际的应用了。 消息队列作为流的典型应用之一， 具有非常好的示范性， 因此我们将使用 Redis 流的相关功能构建一个消息队列应用， 这个消息队列跟我们之前使用其他 Redis 数据结构构建的消息队列具有相似的功能。
代码清单 10-1 展示了一个具有基本功能的消息队列实现：
  代码最开头的是几个转换函数， 它们负责对程序的相关输入输出进行转换和格式化；
  MessageQueue 类用于实现消息队列， 它的添加消息、移除消息以及返回消息数量三个方法分别使用了流的 XADD 命令、 XDEL 命令和 XLEN 命令；
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessa...
于是将序列化方式更改为了Jackson2JsonRedisSerializer
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTe
				一、背景：今天遇到了一个问题，在springboot项目中，想用一个异步的思想完成对数据库的操作，本来想用Mq的方式，但是集成太麻烦了，刚好系统里面已经有了redis，就用了redis做异步的功能。二、需要准备的东西springboot项目，redis能正常用就行【redis集成不会的需要先看其他的文章连接好】三、操作步骤 1、编写一个实体类，其中字段可以根据自己的业务来定
 2、编写一个结果常量类ConstantResult
3、编写一个公共返回结果类CommonResult，用自己系统中的也可以
				一： 说一下当时使用的场景
因为需要用到服务端调用 客户端的场景，这里就无法用到http请求了，因为客户端的pc无固定IP地址，就像手机一样，如果需要推送消息，就必须做任务调度。但是做任务调度的话不停的循环推送消息存在的问题是，
1：需要不停的循环，如果一天只需要推送几次，但是他调用的次数可能成千上万次。
2：循环肯定就有时间间隔，无法实时触发
出于以上两个问题，先后了解到了netty 的长链接机制，mqtt框架的发送机制，都因为太过复杂，并且集成到项目比较困难，所以选择用redis 的发布订阅机制
		//单节点实现方式,如果是单节点建议使用该方式，如果是分布式部署废弃该方式
	    Boolean sendFlag = webMessageHandler.sendMessageToUser(message);
	    if(!sendFlag) {//发送失败广播出去，让其他节点发送
	    	redisTemplate.conver...
stringRedisTemplate.opsForValue().set("test", "100",60*10,TimeUnit.SECONDS);//向redis里存入数据和设置缓存时间
stringRedisTemplate.boundValueOps("test").increment(-1);//val做-1操作
st...
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
2. 在 application.yml 或 application.properties 中配置 Redis 的连接信息：
spring:
  redis:
    host: localhost
    port: 6379
3. 在你的代码中注入 RedisTemplate 对象，并使用它来操作 Redis：
```java
@Autowired
private RedisTemplate<String, Object> redisTemplate;
public void addToDelayQueue(String key, Object value, long delay) {
    redisTemplate.opsForZSet().add(key, value, System.currentTimeMillis() + delay);
4. 使用多线程或者定时任务定期扫描延迟队列，将到期的任务从延迟队列中取出并执行。
```java
@Scheduled(fixedRate = 1000)
public void processDelayQueue() {
    Set<Object> values = redisTemplate.opsForZSet().rangeByScore(DELAY_QUEUE_KEY, 0, System.currentTimeMillis());
    if (!CollectionUtils.isEmpty(values)) {
        redisTemplate.opsForZSet().removeRangeByScore(DELAY_QUEUE_KEY, 0, System.currentTimeMillis());
        values.forEach(value -> {
            // 处理任务
这就是使用 Redis 实现延迟队列的简单示例。你也可以使用其他方式来实现延迟队列，例如使用 Redis 的 list 和 lua 脚本，或者使用基于 Redis 的消息队列框架。