利用Redis 實現消息隊列

時間 2020-01-14

標籤利用 redis 實現消息隊列欄目 Redis 简体版

原文原文鏈接

1.用redis中的List能夠實現隊列，這樣能夠用來作消息處理和任務調度的隊列

2.代碼模擬

代碼結構java

生產者模擬程序redis

/** 
 *  
 */  
package scheduleTest;  
  
import java.util.Random;  
import java.util.UUID;  
  
import redis.clients.jedis.Jedis;  
  
/** 
 * 模擬一個生產者 
 * <p>Title: TaskProducer</p> 
 * <p>Description: </p> 
 * <p>Company: </p> 
 * @author 夏 傑 
 * @date 2015年12月11日 下午4:26:48 
 * @vesion 1.0 
*/  
public class TaskProducer implements Runnable{  
    Jedis jedis = new Jedis("120.55.195.177",6379);  
      
    public void run() {  
        Random random = new Random();  
        while(true){  
            try{  
                Thread.sleep(random.nextInt(600) + 600);  
                // 模擬生成一個任務  
                UUID taskid = UUID.randomUUID();  
                //將任務插入任務隊列：task-queue  
                jedis.lpush("task-queue", taskid.toString());  
                System.out.println("插入了一個新的任務： " + taskid);  
  
            }catch(Exception e){  
                e.printStackTrace();  
            }  
        }  
          
    }  
  
}

消費者模擬程序dom

/** 
 *  
 */  
package scheduleTest;  
  
import java.util.Random;  
  
import redis.clients.jedis.Jedis;  
  
/** 
 * 模擬消費者 
 * <p>Title: TaskConsumer</p> 
 * <p>Description: </p> 
 * <p>Company: </p> 
 * @author 夏 傑 
 * @date 2015年12月11日 下午4:44:23 
 * @vesion 1.0 
*/  
public class TaskConsumer implements Runnable {  
    Jedis jedis = new Jedis("120.55.195.177",6379);  
  
    public void run() {  
        Random random = new Random();  
          
        while(true){  
              
            //從任務隊列"task-queue"中獲取一個任務，並將該任務放入暫存隊列"tmp-queue"  
            String taskid = jedis.rpoplpush("task-queue", "tmp-queue");  
              
              
            // 處理任務----純屬業務邏輯，模擬一下：睡覺  
            try {  
                Thread.sleep(1000);  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            }  
              
              
            //模擬成功和失敗的偶然現象  
            if(random.nextInt(13) % 7 == 0){// 模擬失敗的狀況,機率爲2/13  
                //將本次處理失敗的任務從暫存隊列"tmp-queue"中，彈回任務隊列"task-queue"  
                jedis.rpoplpush("tmp-queue", "task-queue");  
                System.out.println(taskid + "處理失敗，被彈回任務隊列");  
              
            } else {// 模擬成功的狀況  
                  
                // 將本次任務從暫存隊列"tmp-queue"中清除  
                jedis.rpop("tmp-queue");  
                System.out.println(taskid+"處理成功，被清除");  
                  
            }     
        }  
                          
    }  
      
  
      
}

調度主程序ide

/** 
 *  
 */  
package scheduleTest;  
  
/** 
 * <p>Title: TaskShedulerSystem</p> 
 * <p>Description: </p> 
 * <p>Company: </p> 
 * @author 夏 傑 
 * @date 2015年12月11日 下午4:19:09 
 * @vesion 1.0 
*/  
public class TaskShedulerSystem {  
    public static void main(String[] args) throws Exception {  
          
        // 啓動一個生產者線程，模擬任務的產生  
        new Thread(new TaskProducer()).start();  
          
        Thread.sleep(15000);  
          
        //啓動一個線程者線程，模擬任務的處理  
        new Thread(new TaskConsumer()).start();  
          
        //主線程休眠  
        Thread.sleep(Long.MAX_VALUE);  
    }  
}

運行結果spa

插入了一個新的任務： 8025c8b8-f81f-4560-a653-3f339cc371a6線程

插入了一個新的任務： 98dcf980-10d9-4df6-8765-a4873f6b6a743d

插入了一個新的任務： d9636112-4cea-4f49-9f70-88e934aa2a66orm

插入了一個新的任務： 0a8a2799-d672-4444-b53d-74b679559565隊列

插入了一個新的任務： 3948e29e-0217-434a-b7fe-c3be2f0b1073ip

插入了一個新的任務： f06c3a03-83a1-4278-a7c8-61d88afcbadf

插入了一個新的任務： fbf7ed91-b1f1-4713-8490-325d77f951b0

插入了一個新的任務： 99cbf0e4-d981-45ad-88f4-10db1604171e

插入了一個新的任務： ca7cd3cf-9ae3-41a1-b8e1-c8ac9a729255

插入了一個新的任務： 3c954253-d195-4185-b27d-390a6e441eaa

插入了一個新的任務： 4b7f2b4d-c77b-4813-9a63-2b6975cb44a1