springboot集成redis实现消息的订阅与发布

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

package com.yundi.atp.config;

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.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;


@Configuration
public class RedisConfig {
    /**
     * 初始化一个Redis消息监听容器
     * @param connectionFactory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 添加其他配置，如线程池大小等
        return container;
    }

    @Bean
    public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setDefaultSerializer(new StringRedisSerializer());
        return redisTemplate;
    }
}

package com.yundi.atp.constant;

public class ChannelConstant {
    /**
     * 广播通道
     */
    public static final String CHANNEL_GLOBAL_NAME = "channel-global";

    /**
     * 单播通道
     */
    public static final String CHANNEL_SINGLE_NAME = "channel-single";
}

package com.yundi.atp.controller;

import com.yundi.atp.constant.ChannelConstant;
import org.springframework.data.redis.core.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;

import javax.annotation.Resource;


@RequestMapping(value = "base")
@RestController
public class BaseController {
    @Resource
    private RedisTemplate redisTemplate;

    /**
     * 发布广播消息
     *
     * @param msg
     */
    @GetMapping(value = "/publish/{msg}")
    public void sendMsg(@PathVariable(value = "msg") String msg) {
        redisTemplate.convertAndSend(ChannelConstant.CHANNEL_GLOBAL_NAME, msg);
    }
}

package com.yundi.atp.listen;

import com.yundi.atp.constant.ChannelConstant;
import lombok.extern.slf4j.Slf4j;
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.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;


@Slf4j
@Component
public class RedisMessageSubscriber implements MessageListener {
    @Autowired
    private RedisMessageListenerContainer redisMessageListenerContainer;

    /**
     * 订阅消息：将订阅者添加到指定的频道
     */
    @PostConstruct
    public void subscribeToChannel() {
        //广播消息
        redisMessageListenerContainer.addMessageListener(this, new ChannelTopic(ChannelConstant.CHANNEL_GLOBAL_NAME));
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String channel = new String(message.getChannel(), StandardCharsets.UTF_8);
        String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("Received message: " + messageBody + " from channel: " + channel);
    }
}