首页 > 代码库 > 【转】 使用Redis的Pub/Sub来实现类似于JMS的消息持久化

【转】 使用Redis的Pub/Sub来实现类似于JMS的消息持久化

2024-08-28 07:55:07   222人阅读

http://blog.csdn.net/canot/article/details/52040415

 

关于个人对Redis提供的Pub/Sub机制的认识在上一篇博客中涉及到了，也提到了关于如何避免Redis的Pub/Sub的一个最大的缺陷的思路—消息的持久化(http://blog.csdn.net/canot/article/details/51975566)。这篇文章主要是关于其思路(Redis的Pub/Sub的消息持久化)的代码实现:

Pub/Sub机制中最核心的Listener的实现:

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;
import redis.clients.util.RedisInputStream;

public class PubSubListener extends JedisPubSub {

    private String clientId;
    private HandlerRedis handlerRedis;

    // 生成PubSubListener的时候必须制定一个id
    public PubSubListener(String cliendId, Jedis jedis) { 
        this.clientId = cliendId;
        jedis.auth("xxxx");
        handlerRedis = new HandlerRedis(jedis);
    }

    @Override
    public void onMessage(String channel, String message) {
        if ("quit".equals(message)) {
            this.unsubscribe(channel);
        }
        handlerRedis.handler(channel, message);
    }

    // 真正处理接受的地方
    private void message(String channel, String message) {
        System.out.println("message receive:" + message + ",channel:" + channel + "...");
    }

    @Override
    public void onPMessage(String pattern, String channel, String message) {
        // TODO Auto-generated method stub

    }

    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        // 将订阅者保存在一个"订阅活跃者集合中"
        handlerRedis.subscribe(channel);
        System.out.println("subscribe:" + channel);
    }

    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) {
        handlerRedis.ubsubscribe(channel);
        System.out.println("unsubscribe:" + channel);
    }

    @Override
    public void onPUnsubscribe(String pattern, int subscribedChannels) {
        // TODO Auto-generated method stub

    }

    @Override
    public void onPSubscribe(String pattern, int subscribedChannels) {
        // TODO Auto-generated method stub

    }

    class HandlerRedis {
        private Jedis jedis;

        public HandlerRedis(Jedis jedis) {
            this.jedis = jedis;
        }

        public void handler(String channel, String message) {
            int index = message.indexOf("/");
            if (index < 0) {
                // 消息不合法,丢弃
                return;
            }
            Long txid = Long.valueOf(message.substring(0, index));
            String key = clientId + "/" + channel;
            while (true) {
                String lm = jedis.lindex(key, 0);// 获取第一个消息
                if (lm == null) {
                    break;
                }
                int li = lm.indexOf("/");
                // 如果消息不合法，删除并处理
                if (li < 0) {
                    String result = jedis.lpop(key);// 删除当前message
                    // 为空
                    if (result == null) {
                        break;
                    }
                    message(channel, lm);
                    continue;
                }
                Long lxid = Long.valueOf(lm.substring(0, li));// 获取消息的txid
                // 直接消费txid之前的残留消息
                if (txid >= lxid) {
                    jedis.lpop(key);// 删除当前message
                    message(channel, lm);
                    continue;
                } else {
                    break;
                }
            }
        }

        // 持久化订阅操作
        public void subscribe(String channel) {
            // 保证在订阅者集合中的格式为 唯一标识符/订阅的通道
            String key = clientId + "/" + channel;
            // 判断该客户端是否在集合中存在
            boolean isExist = jedis.sismember("PERSIS_SUB", key);
            if (!isExist) {
                // 不存在则添加
                jedis.sadd("PERSIS_SUB", key);
            }
        }

        public void ubsubscribe(String channel) {
            String key = clientId + "/" + channel;
            // 从“活跃订阅者”集合中
            jedis.srem("PERSIS_SUB", key);
            // 删除“订阅者消息队列”
            jedis.del(channel);
        }
    }
}
  • 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
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125

Listener中定义了一个内部类HandlerRedis。Listener类将onMessage以及onSubscribe两个方法交付于HandlerRedis。Handler处理这个方法的时候也即进行着队列的维护。Listener类中定义了一个message()方法，该方法是handler的回调方法，即真正的处理消息的地方。

通道的订阅客户端类:

import redis.clients.jedis.Jedis;

public class SubClient {
   private Jedis jedis;
   private PubSubListener listener;

   public SubClient(String host,PubSubListener pubSubListener){
       jedis = new Jedis(host);
       jedis.auth("XXXXX");
       this.listener = pubSubListener;
   }

   public void sub(String channel){
        jedis.subscribe(listener, channel);
    }

    public void unsubscribe(String channel){
        listener.unsubscribe(channel);
    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

通道的消息发布的客户端:

import java.util.Set;

import redis.clients.jedis.Jedis;

public class PubClient {
    private Jedis jedis;

    public PubClient(String host) {
        jedis = new Jedis(host);
        jedis.auth("wx950709");
    }

    /**
     * 发布的每条消息，都需要在“订阅者消息队列”中持久
     * 
     * @param message
     */
    public void put(String message) {
        //获取所有活跃的消息接收者客户端 clientID/channel
        Set<String> subClients = jedis.smembers("PERSIS_SUB");
        for (String subs : subClients) {
            // 保存每个客户端的消息
            jedis.rpush(subs, message);
        }
    }

    public void publish(String channel, String message) {
        // 每个消息，都有具有一个全局唯一的id
        // txid为了防止订阅端在数据处理时“乱序”，这就要求订阅者需要解析message
        Long txid = jedis.incr("MESSAGE_TXID");
        String content = txid + "/" + message;
        this.put(content);
        jedis.publish(channel, content);//为每个消息设定id，最终消息格式1000/messageContent
    }
    public void close(String channel){
        jedis.publish(channel, "quit");
        jedis.del(channel);//删除
    }
}
  • 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

测试引导类:

import redis.clients.jedis.Jedis;

public class Main {
    public static void main(String[] args) throws Exception{
        PubClient pubClient = new PubClient("127.0.0.1");
        final String channel = "pubsub-channel222";
        PubSubListener listener = new PubSubListener("client_one", new Jedis("127.0.0.1"));
        SubClient subClient = new SubClient("127.0.0.1", listener);
        Thread t1 = new Thread(new Runnable() {

            @Override
            public void run() {
                //在API级别，此处为轮询操作，直到unsubscribe调用，才会返回
                subClient.sub(channel); 
            }

        });
        t1.setDaemon(true);
        t1.start();

        int i = 0;
        while(i < 2){
            pubClient.publish(channel, "message"+i);
            i++;
            Thread.sleep(1000);
        }
    }
}
  • 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
技术分享
 
 

【转】 使用Redis的Pub/Sub来实现类似于JMS的消息持久化


联系
我们