美文网首页
Redis消息队列与订阅者发布者

Redis消息队列与订阅者发布者

作者: 代码potty | 来源:发表于2018-10-16 11:40 被阅读0次

    Redis列表List是采用的双端链表的结构,所有头尾存取元素特别快


    image.png

    手动操作一下在java中实现redis的消息队列,通过生产者和消费者的模式进行实现

    生产者代码:

        public class Producer extends Thread{
            public static final String messageKey = "message";
            private JedisPool jedisPool;
            private String producerName;
            private Jedis jedis;
            private volatile int count = 0;
        
            public Producer(String producerName){
                this.producerName = producerName;
                jedisPool = new JedisPool("127.0.0.1",6379);//redis的ip和端口
                jedis = jedisPool.getResource();
            }
        
        
            public void putMessage(String message){
                jedis.lpush(messageKey,message);
                count++;
            }
        
            public void run(){
                while (true){
                    try {
                        putMessage(RandomStringUtils.randomAlphabetic(5));//获取长度为5的任意字符串
                        TimeUnit.SECONDS.sleep(1);//线程进入为时一秒的休眠状态
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        
            public int getCount(){
                return count;
            }
        
            public static void main(String[] args) throws InterruptedException {
                Producer producer = new Producer("myTest");
                producer.start();
        
                for (;;){
                    System.out.println("生产者:"+producer.producerName+"生产了"+producer.getCount()+"对象");
                    TimeUnit.SECONDS.sleep(10);
                }
            }
        }     
    

    消费者代码

        public class Customer extends Thread{
            private String customerName;
            private volatile int count = 0;
            private Jedis jedis;
            private JedisPool jedisPool;
        
            public Customer(String customerName){
                this.customerName = customerName;
                jedisPool = new JedisPool("127.0.0.1",6379);
                jedis = jedisPool.getResource();
            }
        
            //注意点
            public void getMessage(){
                List<String> result = jedis.blpop(0,Producer.messageKey);
                if (result.size()!=0){
                    String rkey = result.get(0);
                    if (rkey.equals(Producer.messageKey)){
                        String msg = result.get(1);
                        count++;
                        showMessage(msg);
                    }
                }
            }
        
            public void showMessage(String msg){
                System.out.println("消费者:"+customerName+"消息内容是"+msg+"这是第"+count+"个产品");
            }
        
            public void run(){
                while (true){
                    getMessage();
                }
            }
        
            public static void main(String[] args) {
                Customer customer = new Customer("消费一号");
                customer.start();
            }
        
        }
    

    其中比较重要的方法就是getMessage(),可以看到我使用的是blpop()的方式取出消息的,因为blpop是一个阻塞的方法,当消息队列中没有元素,那么就会一直进入到阻塞的状态,如果使用lpop的方式,那么就会一直进行尝试,会浪费很多的资源,得不偿失,所以在此使用拥有阻塞功能的blpop()方法可以提供资源的利用率。


    redis还提供了订阅者和发布者的功能,类似于QQ聊天,可以私聊,群聊的方式。

    发布者代码

        public class Publisher {
            public static final String channel = "aChannel";
            private JedisPool jedisPool;
            private Jedis jedis;
            private String pname;
        
            public Publisher(String pname){
                this.pname = pname;
                jedisPool = new JedisPool("127.0.0.1",6379);
                jedis = jedisPool.getResource();
            }
        
            public void publish(String msg){
                if (org.apache.commons.lang.StringUtils.isBlank(msg))
                    return;
                jedis.publish(channel,msg);
            }
        
            public static void main(String[] args) {
                Publisher publisher = new Publisher("发布者");
                publisher.publish("cececcee");
            }
        } 
    

    这个代码块主要注意在方法publish中,发布消息的时候,消息不能为空,如果传入空值字符串则直接返回。

    订阅者

        public class Subscriber {
            private final String exitInfo = "exit";
            private JedisPool jedisPool;
            private Jedis jedis;
            private String subscriber;
        
            public Subscriber(String subscriber){
                this.subscriber = subscriber;
                jedisPool = new JedisPool("127.0.0.1",6379);
                jedis = jedisPool.getResource();
            }
        
            public void subscribe(String ...channel){
                if (channel==null || channel.length<=0)
                    return;
        
                JedisPubSub jedisPubSub = new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        if (Publisher.channel.equals(channel)){
                            System.out.println("接收到频道:"+channel+"的消息:"+message);
                            if (message.equals(exitInfo)){
                                System.exit(0);
                            }
                        }
                    }
        
                    @Override
                    public void onSubscribe(String channel, int subscribedChannels) {
                        if (Publisher.channel.equals(channel)){
                            System.out.println("订阅了频道:"+channel);
                        }
                    }
                };
                jedis.subscribe(jedisPubSub,channel);
            }
        
            public static void main(String[] args) {
                Subscriber subscriber = new Subscriber("订阅者");
                subscriber.subscribe(Publisher.channel);
            }
        }       
    

    订阅者中,主要注意JedisPubSub这个类,JedisPubSub类是一个没有抽象方法的抽象类,里面方法都是一些空实现,所以我们可以选择性的去重写某些方法来实现我们的目的,在代码块中可以看出我重写了onMessage()方法和onSubscribe()方法,前一个方法是在接收数据时候的处理,后一个方法是在订阅频道时候的处理,最后再调用jedis的subscribe()方法来实现订阅频道。


    image.png
    image.png

    经过测试,成功发送和接收信息。

    总结

    使用Redis的List数据结构可以简单迅速地做一个消息队列,同时Redis提供的BRPOP和BLPOP等指令解决了频繁调用Jedis的rpop和lpop方法造成的资源浪费问题。除此之外,Redis提供对发布/订阅模式的指令,可以实现消息传递、进程间通信。

    补充说明随机字符串实现类

        //产生5位长度的随机字符串,中文环境下是乱码
        RandomStringUtils.random(5);
    
        //使用指定的字符生成5位长度的随机字符串
        RandomStringUtils.random(5, new char[]{'a','b','c','d','e','f', '1', '2', '3'});
        
        //生成指定长度的字母和数字的随机组合字符串
        RandomStringUtils.randomAlphanumeric(5);
    
        //生成随机数字字符串
        RandomStringUtils.randomNumeric(5);
        
        //生成随机[a-z]字符串,包含大小写
        RandomStringUtils.randomAlphabetic(5);
    
        //生成从ASCII 32到126组成的随机字符串 
        RandomStringUtils.randomAscii(4)
    

    想要使用这个类的方法需要导入相应的包

        <dependency>
            <groupId>org.apache.directory.studio</groupId>
            <artifactId>org.apache.commons.lang</artifactId>
            <version>2.6</version>
        </dependency>
    

    参考链接:https://blog.csdn.net/qq_34212276/article/details/78455004

    相关文章

      网友评论

          本文标题:Redis消息队列与订阅者发布者

          本文链接:https://www.haomeiwen.com/subject/afoczftx.html