侧边栏壁纸
博主头像
IILee博主等级

所以,看不到光,算是不幸吗?需要光才是真正的不幸吧。

  • 累计撰写 89 篇文章
  • 累计创建 63 个标签
  • 累计收到 26 条评论
隐藏侧边栏

Spring Boot RedisTemplate 实现发布/订阅

IILee
2021-07-01 / 0 评论 / 7 点赞 / 142 阅读 / 4,435 字 / 正在检测是否收录...
温馨提示:
本文最后更新于 2021-07-04,若内容或图片失效,请留言反馈。部分素材来自网络,若不小心影响到您的利益,请联系我们删除。

前言

Redis 本身支持消息的发布与订阅,本文讲解Spring Boot项目中如何实现消息发布/订阅。

内容

  • 方式一

配置消息:

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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;

/**
 * @author IILee
 */
@Configuration
public class RedisPubSubConfig {

    /**
     * 初始化监听器
     * @param connectionFactory
     * @param listenerAdapter
     * @return
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,
                                                   MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, new PatternTopic("demo-channels"));
        return container;
    }

    @Bean
    public MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(proxyMessageReceiver(), "process");
    }

    @Bean
    public ProxyMessageReceiver proxyMessageReceiver() {
        return new ProxyMessageReceiver();
    }
}

消费者如下:

public class ProxyMessageReceiver {
    private final Logger LOG = LoggerFactory.getLogger(this.getClass());

    @Resource
    private MessageProcessService messageProcessService;

    public void process(String body) {

    }

}
  • 方式二

配置如下:

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

/**
 * @author IILee
 */
@Configuration
public class RedisPubSubConfig {

    /**
     * 初始化监听器
     * @param connectionFactory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(proxyMessageReceiver(), new PatternTopic("demo-channels"));
        return container;
    }

    @Bean
    public ProxyMessageReceiver proxyMessageReceiver() {
        return new ProxyMessageReceiver();
    }

}

消费者代码如下:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

import javax.annotation.Resource;

/**
 * 下游消息处理类(包括:用户消息下发、关闭会话连接等)
 * @author IILee
 */
public class ProxyMessageReceiver implements MessageListener {
    private final Logger LOG = LoggerFactory.getLogger(this.getClass());

    @Resource
    private MessageProcessService messageProcessService;

    @Override
    public void onMessage(Message message, byte[] pattern) {

        String traceId = TracingContext.genIfAbsentTraceId();
        //trace
        TracingContext.putTraceId(traceId);

        try {
            String channel = StringUtils.getStringUtf8(message.getChannel());
            String body = StringUtils.getStringUtf8(message.getBody());

            LOG.info("客服呼叫中心-处理下游消息开始, channel:{}, traceId:{}, 请求内容:{}", channel, traceId, body);

            CallCentreMessageDTO messageDTO = JsonUtils.parseObject(body, CallCentreMessageDTO.class);

            boolean success = messageProcessService.process(messageDTO);
            LOG.info("客服呼叫中心-处理下游消息结束, traceId:{} 处理结果:{}", traceId, success);
        } catch (Exception e) {
            LOG.error(String.format("客服呼叫中心-处理下游消息异常, messageId:%s", traceId), e);
        } finally {
            TracingContext.removeTraceId();
        }
    }
}

  • 消费者代码
@Service
public class RedisService {
    @Resource
    private StringRedisTemplate stringRedisTemplate;

  //向通道发送消息的方法
    public void sendChannelMess(String channel, String message) {
        stringRedisTemplate.convertAndSend(channel, message);
    }
}

后记

自测示例代码,生产环境需更详尽处理。

0

评论