springboot项目集成redis的订阅发布_tioclustertopic

前面文章写过jedis的工具类，这篇文章主要将redis的发布订阅整理出来。
（使用jedis操作redis）

0. 整体结构

1. 引入依赖

		<dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
1
2
3
4
5
6
7
8
9

2. 定义发布渠道-实体类

package com.hzdf.config.redis;

/**
 * @program: AuctionPlatform
 *
 * @description: 发布定义
 *
 * @author: Mr.Wei
 *
 * @create: 2022-06-23 14:04
 **/

public interface RedisPubSub {

    /**
     * 订阅处理方法
     */
    String INVOKE_METHOD = "receiveMessage";


    /**
     * 订阅频道:拍卖业务
     */
    String STUDENT_RECORD = "applet:auction";


}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

3. 发布消息

package com.hzdf.config.redis;

/**
 * @program: AuctionPlatform
 *
 * @description: 发布消息
 *
 * @author: Mr.Wei
 *
 * @create: 2022-06-23 14:08
 **/

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import static com.hzdf.ap.util.ExceptionUtil.tryError;
import static com.hzdf.ap.util.RedisUtils.returnResource;

@Service
public class RedisPublisher {

    @Autowired
    private JedisPool jedisPool;

    public void sendMsg(String channel, String msg){
        //从 mychannel 的频道上推送消息
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.publish(channel, msg);
        } catch (Exception e) {
            tryError("redis执行-RedisPublisher-异常:", e);
        } finally {
            returnResource(jedisPool, jedis);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

4. 订阅消息（接收）

package com.hzdf.config.redis;

import org.springframework.beans.factory.annotation.Autowired;
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;

import java.util.ArrayList;
import java.util.List;

/**
 * @program: AuctionPlatform
 *
 * @description: 配置消息订阅
 *
 * @author: Mr.Wei
 *
 * @create: 2022-06-23 14:03
 **/
@Configuration
public class RedisListener {

    @Autowired
    private BusinessSubService businessSubService;

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                   MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        List<PatternTopic> topicList = new ArrayList<>();
        topicList.add(new PatternTopic(RedisPubSub.STUDENT_RECORD));
        container.addMessageListener(listenerAdapter, topicList);
        return container;
    }

    @Bean
    public MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(businessSubService, RedisPubSub.INVOKE_METHOD);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

5. 业务处理

package com.hzdf.config.redis;/**
 * @program: AuctionPlatform
 * @description: 业务处理
 * @author: Mr.Wei
 * @create: 2022-06-23 14:06
 **/

/**
 * @program: AuctionPlatform
 *
 * @description: 业务处理
 *
 * @author: Mr.Wei
 *
 * @create: 2022-06-23 14:06
 **/

import com.alibaba.fastjson.JSONObject;
import com.hzdf.ap.controller.WebSocketServer;
import com.hzdf.ap.controller.WebSocketServerAuction;
import com.hzdf.tio.MyTioClusterTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.tio.cluster.TioClusterVo;

import java.util.Map;

import static com.hzdf.ap.util.ExceptionUtil.tryError;

@Service
public class BusinessSubService {

    private static Logger log = LoggerFactory.getLogger(BusinessSubService.class);

    public void receiveMessage(String message, String channel) {
        log.info("################SUBSCRIBE######### message:{}, channel:{} #############", message, channel);
        switch (channel) {
            case RedisPubSub.STUDENT_RECORD:
                msgSubService(message);
                break;
            default:
                break;
        }
    }

    /**
     * 解析消息体进行业务操作
     * socket推送
     * @param message
     */
    public void msgSubService(String message){
        if(StringUtils.isEmpty(message)){
            log.warn("接收到订阅消息为空-msgSubService,频道为:" + RedisPubSub.STUDENT_RECORD);
            return;
        }
        
        //以下是业务操作
        System.out.println("接收到消息" + message);
        try {
        	//这里调用自己写的业务方法即可
        	//这里调用自己写的业务方法即可
        	//这里调用自己写的业务方法即可
            MyTioClusterTopic.receiveMsg(message);
        
        }catch (Exception e){
            tryError("接收到订阅消息-解析消息体-异常",e);
        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

6. 工具类(tryError)

该工具类也是为了打印Error日志，直接使用log.error();同理。
1

package com.hzdf.ap.util;/**
 * @program:
 * @author: zxy
 * @create: 2021-04-30 13:58
 **/

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;

/**
 * @program:
 *
 * @description:异常信息整体处理到日志中
 *
 * @author: zxy
 *
 * @create: 2021-04-30 13:58
 **/
public class ExceptionUtil {
    private static final Logger logger = LoggerFactory.getLogger(ExceptionUtil.class);
    public static void tryError(String mag, Exception e) {
        StringWriter sw = null;

        PrintWriter pw = null;

        try {
            sw = new StringWriter();

            pw = new PrintWriter(sw);

// 将出错的栈信息输出到printWriter中

            e.printStackTrace(pw);

            pw.flush();

            sw.flush();

        } finally {
            if (sw != null) {
                try {
                    sw.close();

                } catch (IOException e1) {
                    e1.printStackTrace();

                }

            }

            if (pw != null) {
                pw.close();

            }

        }
         logger.error(mag+":####:"+sw.toString());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

7. 使用说明

发布消息：直接注入 RedisPublisher，调用sengMsg即可。

	@Autowired
    private RedisPublisher redisPublisher;

	public void publish() {
        redisPublisher.sendMsg("applet:auction","我是要发送的消息");
    }
1
2
3
4
5
6

订阅消息：将 BusinessSubService{} 中的MyTioClusterTopic.receiveMsg(message);方法直接替换为自己的业务代码即可。