redis队列

Posted 2022-04-09 enhance

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了redis队列相关的知识，希望对你有一定的参考价值。

1、lpush+rpop

　　采用rpop需要不停调用rpop方法查看list中是否有待处理消息。每调用一次都会发起一次连接，造成不必要浪费

技术图片

　　代码：

　　　　　　producer：　

package com.eval.mind.service.redis;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

import redis.clients.jedis.Jedis;
public class Producer_Lpush extends Thread
    
    public static final String MESSAGE_KEY = "message:queue";
    private Jedis jedis;
    private String producerName;
    private volatile int count;
 
    public Producer_Lpush(String name) 
        this.producerName = name;
        init();
    
 
    private void init() 
        jedis=new Jedis("192.168.80.4",6379);
    
    
    public void putMessage(String message) 
        Long size = jedis.lpush(MESSAGE_KEY, message);
        System.out.println(producerName + ": 当前未被处理消息条数为:" + size);
        count++;
    
 
    public int getCount() 
        return count;
    
    
    @Override
    public void run() 
        try 
            while (true) 
                putMessage(UUID.randomUUID().toString());
                TimeUnit.SECONDS.sleep(1);
            
         catch (InterruptedException e) 
 
         catch (Exception e) 
            e.printStackTrace();
        
    
 
    public static void main(String[] args) throws InterruptedException
        Producer_Lpush producer = new Producer_Lpush("myProducer");
        producer.start();
 
        for(; ;) 
            System.out.println("main : 已存储消息条数:" + producer.getCount());
            TimeUnit.SECONDS.sleep(10);

View Code

　　　　　consumer:

package com.eval.mind.service.redis;

import redis.clients.jedis.Jedis;

/*
 * rpop从redis队列中pop处一个String类型元素，String mes=jedis.rpop(key)
 * 缺点是消费端需要不停的调用rpop方法查看list是否有待处理消息。每调用依次都会发起依次连接，这会造成不必要浪费；为解决此参考Consumer_Brpop.java方法
 */
public class Consumer_Rpop extends Thread 
    private String customerName;
    private volatile int count;
    private Jedis jedis;

    public Consumer_Rpop(String name) 
            this.customerName = name;
            init();
        

    private void init() 
        jedis = new Jedis("192.168.80.4", 6379);
    

    public void processMessage() 
        String message = jedis.rpop(Producer_Lpush.MESSAGE_KEY); //如果redis队列中没有数据，也会一直调用
        if (message != null) 
            count++;
            handle(message);
        
    

    public void handle(String message) 
        System.out.println(customerName + " 正在处理消息,消息内容是: " + message + " 这是第" + count + "条");
    

    @Override
    public void run() 
        while (true) 
            processMessage();
        
    

    public static void main(String[] args) 
        Consumer_Rpop customer = new Consumer_Rpop("yamikaze");
        customer.start();

View Code

2、lpush+brpop

　　brpop：blocking rpop，采用brpop时，如果redis队列中不存在数据则调用List<String> messages=jedis.brpop(int i,key1,key2)时会阻塞，不会往下执行，一直等待redis队列中再次push进数据后继续执行pop操作

　 * brpop支持多个列表(队列)

   * brpop指令是支持队列优先级的，比如这个例子中MESSAGE_KEY的优先级大于testKey（依据brpop中顺序决定）。
   * 如果两个列表中都有元素，会优先返回优先级高的列表中的元素，所以这儿优先返回MESSAGE_KEY
   * 0表示不限制等待，会一直阻塞在这儿

　　代码：

　　　　　　producer：　　

package com.eval.mind.service.redis;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

import redis.clients.jedis.Jedis;
public class Producer_Lpush extends Thread
    
    public static final String MESSAGE_KEY = "message:queue";
    private Jedis jedis;
    private String producerName;
    private volatile int count;
 
    public Producer_Lpush(String name) 
        this.producerName = name;
        init();
    
 
    private void init() 
        jedis=new Jedis("192.168.80.4",6379);
    
    
    public void putMessage(String message) 
        Long size = jedis.lpush(MESSAGE_KEY, message);
        System.out.println(producerName + ": 当前未被处理消息条数为:" + size);
        count++;
    
 
    public int getCount() 
        return count;
    
    
    @Override
    public void run() 
        try 
            while (true) 
                putMessage(UUID.randomUUID().toString());
                TimeUnit.SECONDS.sleep(1);
            
         catch (InterruptedException e) 
 
         catch (Exception e) 
            e.printStackTrace();
        
    
 
    public static void main(String[] args) throws InterruptedException
        Producer_Lpush producer = new Producer_Lpush("myProducer");
        producer.start();
 
        for(; ;) 
            System.out.println("main : 已存储消息条数:" + producer.getCount());
            TimeUnit.SECONDS.sleep(10);

View Code

　　　　　 consumer：　　　

package com.eval.mind.service.redis;

import java.util.List;

import org.springframework.util.CollectionUtils;

import redis.clients.jedis.Jedis;

/*
 * brpop:blocking rpop,这个指令只有在有元素时才返回，没有则会阻塞到超时返回null；
 * brpop支持多个列表（队列），支持队列优先级，比如messagekey优先级大于testkey，如果两个列表中都有元素，会优先返回messagekey列表中元素
 * 返回List<String>类型  List<String> messages=jedis.brpop(key);
 */
public class Consumer_Brpop extends Thread
    private volatile int count;
    String name;
    private Jedis jedis;
    public Consumer_Brpop(String name) 
        this.name=name;
        init();
    
    private void init() 
        jedis = new Jedis("192.168.80.4", 6379);
    
    
    private void processMessage() 
        List<String> messages=jedis.brpop(0, "message:queue","message:tmp"); //如果redis队列中没有数据，此处会阻塞不会往下执行，直到redis队列又push进数据
        if(!CollectionUtils.isEmpty(messages)) 
            String keyName=messages.get(0);
            String messageValue=messages.get(1);
            count++;
            handle(messageValue);
        
    
    
    public void handle(String message) 
        System.out.println(name + " 正在处理消息,消息内容是: " + message + " 这是第" + count + "条");
    
    
    @Override
    public void run() 
        while (true) 
            processMessage();
        
    
    
    public static void main(String[] args) 
        Consumer_Brpop customer = new Consumer_Brpop("yamikaze");
        customer.start();

View Code

3、publish+subscribe

　　　redis除了对消息队列提供支持外，还提供一组命令用于支持发布/订阅模式

　　　发布：

　　　　　　publish指令可用于发布一条消息：publish channel message

　　　　　　subscribe指令用于接收一条消息：subscribe channel

　　可以看到使用subscribe指令进入订阅模式后，并没有接收到publish发送的消息，这是因为只有在消息发送出去前才会收到，也就是说订阅subscribe启动要在　　　　 publish之前执行

　　　订阅发布模式和消息队列模式区别：消息队列模式是通过key方式实现，取出就删除了，其他进程取不到。订阅发布可以支持多客户端获取同一个频道（channel）发布的消息

　　代码：

　　　　publish：　　

package com.eval.mind.service.redis;

import org.apache.commons.lang3.StringUtils;

import redis.clients.jedis.Jedis;

public class Producer_Publish 
    public static final String Channel_key="channel:message";
    private Jedis jedis;
    
    public Producer_Publish() 
        jedis=new Jedis("192.168.80.4",6379);
    
    
    public void publishMessage(String message) 
        if(StringUtils.isNoneBlank(message)) 
            return;
        
        jedis.publish(Channel_key, message); //publish方法 发布消息
    
    
    public static void main(String[] args) 
        Producer_Publish publisher=new Producer_Publish();
        publisher.publishMessage("Hello Publish Redis");

View Code

　　　　subscribe：　　　

package com.eval.mind.service.redis;

import org.apache.commons.lang3.StringUtils;

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

public class Consumer_Subscribe 

    private Jedis jedis;
    private static final String EXIT_COMMAND = "exit";

    public Consumer_Subscribe() 
        jedis = new Jedis("192.168.80.4", 6379);
    

    public void subscribe(String channel)  // subscribe方法订阅消息
        if (StringUtils.isBlank(channel)) 
            return;
        
        // JedisPubSub类是一个没有抽象方法的抽象类，里边方法时一些空实现，可以选择需要的方法覆盖，这里使用的是subscribe指令，所以覆盖了onMessage
        // 如果使用pubsubscribe指令则覆盖onPmessage方法

        JedisPubSub jps = new JedisPubSub() 

            @Override
            public void onMessage(String channel, String message) 
                while (true)        //相对于redis队列只能消费一次，此处会对channel_key一直订阅，
                    if (Producer_Publish.Channel_key.equals(channel)) 
                        System.out.println("接收到消息: channel : " + message);
                        // 接收到exit消息后退出
                        if (EXIT_COMMAND.equals(message)) 
                            System.exit(0);
                        
                    
                
            

            /**
             * 订阅时
             */
            @Override
            public void onSubscribe(String channel, int subscribedChannels) 
                if (Producer_Publish.Channel_key.equals(channel)) 
                    System.out.println("订阅了频道:" + channel);
                
            
        ;
        jedis.subscribe(jps, channel);
    

    public static void main(String[] args) 
        Consumer_Subscribe client = new Consumer_Subscribe();
        client.subscribe(Producer_Publish.Channel_key);

View Code

以上是关于redis队列的主要内容，如果未能解决你的问题，请参考以下文章