Redis(十八)：Redis和队列详解大数据

package org.yamikaze.redis.messsage.queue; 
  
import org.yamikaze.redis.test.MyJedisFactory; 
import redis.clients.jedis.Jedis; 
  
import java.util.concurrent.TimeUnit; 
  
/** 
 * 消息生产者 
 * @author yamikaze 
 */ 
public class Producer extends Thread { 
  
    public static final String MESSAGE_KEY = "message:queue"; 
    private Jedis jedis; 
    private String producerName; 
    private volatile int count; 
  
    public Producer(String name) { 
        this.producerName = name; 
        init(); 
    } 
  
    private void init() { 
        jedis = MyJedisFactory.getLocalJedis(); 
    } 
  
    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(StringUtils.generate32Str()); 
                TimeUnit.SECONDS.sleep(1); 
            } 
        } catch (InterruptedException e) { 
  
        } catch (Exception e) { 
            e.printStackTrace(); 
        } 
    } 
  
    public static void main(String[] args) throws InterruptedException{ 
        Producer producer = new Producer("myProducer"); 
        producer.start(); 
  
        for(; ;) { 
            System.out.println("main : 已存储消息条数:" + producer.getCount()); 
            TimeUnit.SECONDS.sleep(10); 
        } 
    } 
}

package org.yamikaze.redis.messsage.queue; 
  
import org.yamikaze.redis.test.MyJedisFactory; 
import redis.clients.jedis.Jedis; 
  
/** 
 * 消息消费者 
 * @author yamikaze 
 */ 
public class Customer extends Thread{ 
  
    private String customerName; 
    private volatile int count; 
    private Jedis jedis; 
  
    public Customer(String name) { 
        this.customerName = name; 
        init(); 
    } 
  
    private void init() { 
        jedis = MyJedisFactory.getLocalJedis(); 
    } 
  
    public void processMessage() { 
        String message = jedis.rpop(Producer.MESSAGE_KEY); 
        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) { 
        Customer customer = new Customer("yamikaze"); 
        customer.start(); 
    } 
}

public void processMessage() { 
    /** 
     * brpop支持多个列表(队列) 
     * brpop指令是支持队列优先级的，比如这个例子中MESSAGE_KEY的优先级大于testKey（顺序决定）。 
     * 如果两个列表中都有元素，会优先返回优先级高的列表中的元素，所以这儿优先返回MESSAGE_KEY 
     * 0表示不限制等待，会一直阻塞在这儿 
     */ 
    List<String> messages = jedis.brpop(0, Producer.MESSAGE_KEY, "testKey"); 
    if(messages.size() != 0) { 
        //由于该指令可以监听多个Key,所以返回的是一个列表 
        //列表由2项组成，1) 列表名，2)数据 
        String keyName = messages.get(0); 
        //如果返回的是MESSAGE_KEY的消息 
        if(Producer.MESSAGE_KEY.equals(keyName)) { 
            String message = messages.get(1); 
            handle(message); 
        } 
  
    } 
    System.out.println("======================="); 
}

package org.yamikaze.redis.messsage.subscribe; 
  
import org.yamikaze.redis.messsage.queue.StringUtils; 
import org.yamikaze.redis.test.MyJedisFactory; 
import redis.clients.jedis.Jedis; 
  
/** 
 * 消息发布方 
 * @author yamikaze 
 */ 
public class Publisher { 
  
    public static final String CHANNEL_KEY = "channel:message"; 
    private Jedis jedis; 
  
    public Publisher() { 
        jedis = MyJedisFactory.getLocalJedis(); 
    } 
  
    public void publishMessage(String message) { 
        if(StringUtils.isBlank(message)) { 
            return; 
        } 
        jedis.publish(CHANNEL_KEY, message); 
    } 
  
    public static void main(String[] args) { 
        Publisher publisher = new Publisher(); 
        publisher.publishMessage("Hello Redis!"); 
    } 
}