基于Redis实现订阅发布功能_Redis

背景

业务发展过程中，希望做到异步解耦，但是又不想引入mq中间件，在中小型服务中，就可以考虑使用redis自带的订阅发布来解决这个问题。使用 redis 实现消息的订阅和发布时，可以通过 spring boot 集成 redis 来方便地实现。

引入redis依赖

<dependency>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-data-redis</artifactid>
</dependency>

配置 redis

在 application.properties 文件中，添加 redis 配置：

spring.redis.host=localhost
spring.redis.port=6379

编写代码

redis 配置

创建一个配置类来配置 redis 的连接工厂和监听器：

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.channeltopic;
import org.springframework.data.redis.listener.redismessagelistenercontainer;
import org.springframework.data.redis.listener.adapter.messagelisteneradapter;
import org.springframework.data.redis.core.stringredistemplate;
@configuration
public class redisconfig {
    @bean
    public redismessagelistenercontainer container(redisconnectionfactory connectionfactory,
                                                   messagelisteneradapter listeneradapter) {
        redismessagelistenercontainer container = new redismessagelistenercontainer();
        container.setconnectionfactory(connectionfactory);
        container.addmessagelistener(listeneradapter, topic());
        return container;
    }
    @bean
    public messagelisteneradapter listeneradapter(redismessagesubscriber subscriber) {
        return new messagelisteneradapter(subscriber, "onmessage");
    }
    @bean
    public channeltopic topic() {
        return new channeltopic("messagequeue");
    }
    @bean
    public stringredistemplate template(redisconnectionfactory connectionfactory) {
        return new stringredistemplate(connectionfactory);
    }
}

创建消息订阅者

编写一个类来处理收到的消息：

import org.springframework.stereotype.service;
@service
public class redismessagesubscriber {
    public void onmessage(string message, string channel) {
        system.out.println("received message: " + message + " from channel: " + channel);
    }
}

创建消息发布者

编写一个发布者通过 redis 模板发送消息：

import org.springframework.beans.factory.annotation.autowired;
import org.springframework.data.redis.core.stringredistemplate;
import org.springframework.data.redis.listener.channeltopic;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.requestparam;
import org.springframework.web.bind.annotation.restcontroller;
@restcontroller
public class messagepublisher {
    @autowired
    private stringredistemplate template;
    @autowired
    private channeltopic topic;
    @getmapping("/publish")
    public string publish(@requestparam string message) {
        template.convertandsend(topic.gettopic(), message);
        return "message published: " + message;
    }
}

如果需要监听多个channel，可以通过redisconfig中添加新的消息适配器。

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.channeltopic;
import org.springframework.data.redis.listener.redismessagelistenercontainer;
import org.springframework.data.redis.listener.adapter.messagelisteneradapter;
import org.springframework.data.redis.core.stringredistemplate;
@configuration
public class redisconfig {
    @bean
    public redismessagelistenercontainer container(redisconnectionfactory connectionfactory,
            messagelisteneradapter listeneradapter1,
            messagelisteneradapter listeneradapter2) {
        redismessagelistenercontainer container = new redismessagelistenercontainer();
        container.setconnectionfactory(connectionfactory);
        container.addmessagelistener(listeneradapter1, topic1());
        container.addmessagelistener(listeneradapter2, topic2());
        return container;
    }
    @bean
    public messagelisteneradapter listeneradapter1(redismessagesubscriber subscriber) {
        return new messagelisteneradapter(subscriber, "onmessage");
    }
    @bean
    public messagelisteneradapter listeneradapter2(redismessagesubscriber subscriber) {
        return new messagelisteneradapter(subscriber, "onmessage");
    }
    @bean
    public channeltopic topic1() {
        return new channeltopic("channelone");
    }
    @bean
    public channeltopic topic2() {
        return new channeltopic("channeltwo");
    }
    @bean
    public stringredistemplate template(redisconnectionfactory connectionfactory) {
        return new stringredistemplate(connectionfactory);
    }
}

同时redismessagesubscriber 也可以书写多个来区分不同的业务场景下不同业务处理。

到此这篇关于redis分布式系统的原理与实操的文章就介绍到这了,更多相关redis分布式系统内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！