为什么要有队列

生产者和消费者的速度不一致，为了让他们可以配合工作，才要使用队列

实现过程

这里用redis实现了简单队列（不阻塞，不可靠，队列大小不固定）不可靠的原因见分析

生产者

package cn.itcast.core.service.redis;

import java.util.UUID;
import redis.clients.jedis.Jedis;

public class Producer implements Runnable {

    Jedis jedis = new Jedis("192.168.254.137", 6379);

    @Override
    public void run() {
        while (true) {
            try {
                Thread.sleep(6000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String taskid = UUID.randomUUID().toString().replaceAll("-", "");
            jedis.lpush("mqueue", taskid);
            System.out.println("插入一个新的任务： " + taskid);
        }
    }

}

消费者

package cn.itcast.core.service.redis;

import java.util.Random;

import redis.clients.jedis.Jedis;

public class Consumer implements Runnable{

    Jedis jedis = new Jedis("192.168.254.137",6379);

    @Override
    public void run() {
         while(true){  
                String taskid = jedis.rpoplpush("mqueue", "tqueue");  
                if(null != taskid){
                    try {
                        Thread.sleep(new Random().nextInt(10000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //处理业务逻辑
                    if( new Random().nextInt(9) % 3 == 0){ 
                        //失败
                        jedis.rpoplpush("tqueue", "mqueue");  
                        System.out.println(taskid + "处理失败，重新放入队列");  

                    } else {
                        //成功
                        jedis.rpop("tqueue");  
                        System.out.println(taskid+"处理成功");  

                    }     
                }

            }  

    } 

}

测试

package cn.itcast.core.service.redis;

public class TestQueue {

    public static void main(String[] args) throws InterruptedException {

        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();

        Thread.sleep(Integer.MAX_VALUE);
    }
}

测试结果

redis 里的数据

分析

这里的tqueue还保留一条数据的原因是 当消费者和生产者线程都挂掉时候，在消费者中 String taskid = jedis.rpoplpush(“mqueue”, “tqueue”); 代码后面，线程休息， jedis.rpop(“tqueue”);没有及时执行 ，所以tqueue有一条数据。重新运行消费者线程时候，这三处的代码将会使数据不一致
String taskid = jedis.rpoplpush(“mqueue”, “tqueue”);
jedis.rpoplpush(“tqueue”, “mqueue”);
jedis.rpop(“tqueue”);

所以直接使用上面的代码是不适合的。