SpringBoot2.X 整合RedisTemplate 简单实现消息队列

首先：SpringBoot2 以上 整合redis与 Springboot1 有所区别，不用配置redis

在启动的时候，容器中会根据application中redis的配置自动配置，可在项目里直接引用RedisTemplate

下面是SpringBoot引用Redis的pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>demo</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!--整合redis-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

下面实现消息队列：

准备两个SpringBoot项目

①发布消息application：

server:
  port: 8010
spring:
  redis:
    database: 0
    host: 127.0.0.1
    port: 6378
    timeout: 20000

②只是一个demo 没有service解耦，下面是controller

package com.example.demo.redis.controller;

import com.example.demo.entity.User;
import com.example.demo.redis.service.PubRedisService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;

@RequestMapping("/redis")
@Controller
public class RedisController {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private PubRedisService pubRedisService;
    @RequestMapping("/index")
    public User redisIndex() {
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        ops.set("user", "user2");
        String str = ops.get("a");
        System.out.println("redis server str:" + str);
        return null;
    }

    /**
     * 发布消息
     * @param id
     * @return
     */
    @RequestMapping("/sendMessage/{id}")
    public String sendMessage(@PathVariable String id) {
        redisTemplate.convertAndSend("msg","哈哈哈，redis 订阅信息");
        return "";
    }
}

消息发布就完事了

接下来是接收消息

server:
  port: 8010
spring:
  redis:
    database: 0
    host: 127.0.0.1
    port: 6378
    timeout: 20000

然后是接收消息的配置

新建一个接收消息的实体类

package com.example.demo.redis.entity;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

/**
 * 接收消息的实体类
 */
@Component
public class RedisMessage implements MessageListener {
    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;
    @Override
    public void onMessage(Message message, byte[] pattern) {
        RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
        String msg = serializer.deserialize(message.getBody());
        System.out.println("接收到的消息是：" + msg);
    }
}

配置消息adapter

package com.example.demo.redis.config;


import com.example.demo.redis.entity.RedisMessage;
import com.sun.istack.internal.tools.DefaultAuthenticator;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
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;

import java.util.concurrent.CountDownLatch;

/**
 * 消息队列  订阅者  redis配置
 */
@Configuration
//@AutoConfigureAfter({RedisMessage.class})
public class RedisSubConfig {

    /**
     * 创建连接工厂
     *
     * @param connectionFactory
     * @param adapter
     * @return
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                   MessageListenerAdapter adapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(adapter, new PatternTopic("msg"));
        return container;
    }

    /**
     * @param message
     * @return
     */
    @Bean
    public MessageListenerAdapter adapter(RedisMessage message){
        // onMessage 如果RedisMessage 中 没有实现接口，这个参数必须跟RedisMessage中的读取信息的方法名称一样
        return new MessageListenerAdapter(message, "onMessage");
    }
}

消息接收者已经定义好了

启动redis server 和两个SpringBoot项目  访问发布者的接口发布消息，消息接收者就会自动接收发布者发布的消息，这个消息队列实时性很高