【並發】8、借助redis 實現多線程生產消費阻塞隊列

本文轉載自查看原文 2019-06-12 17:37 1937 並發

顧名思義這個就是再消費的時候，不是之前的那哥用yield進行線程切換的操作，而是用線程等待阻塞的方式去執行，說實話我感覺效率不一定有之前那個好，

因為我對這種阻塞隊列使用的時候，之前有發現阻塞隊列，塞着塞着線程就會進入假死狀態，這個很奇怪，但是有的時候又是好的，這個也不清楚到底是為什么

但是畢竟也是一種實現，我就寫出來了看看吧

生產者

package queue.redisQueue;

import queue.fqueue.vo.TempVo;
import redis.clients.jedis.Jedis;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.util.UUID;

/**
 * @ProjectName: cutter-point
 * @Package: queue.redisQueue
 * @ClassName: RedisQueueProducter2
 * @Author: xiaof
 * @Description: ${description}
 * @Date: 2019/6/12 16:29
 * @Version: 1.0
 */
public class RedisQueueProducter2 implements Runnable {

    private Jedis jedis;
    private String queueKey;

    public RedisQueueProducter2(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    @Override
    public void run() {

        while(true) {

            try {
                Thread.sleep((long) (Math.random() * 1000));

                //不存在則創建，存在則直接插入
                //向redis隊列中存放數據
                //生成數據
                TempVo tempVo = new TempVo();
                tempVo.setName(Thread.currentThread().getName() + ",time is:" + UUID.randomUUID());
                //序列化為字節
                ByteArrayOutputStream arrayOutputStream = new ByteArrayOutputStream();
                ObjectOutputStream objectOutputStream = new ObjectOutputStream(arrayOutputStream);
                objectOutputStream.writeObject(tempVo);
                arrayOutputStream.flush();

                try {
                    int i = 0;
                    while(i < 10) {
                        long num = jedis.lpush(queueKey.getBytes(), arrayOutputStream.toByteArray());
                        if(num > 0) {
                            System.out.println("成功！");
                            break;
                        }

                        ++i;
                    }
                } catch (Exception e) {
                    System.out.println("失敗！");
//                    long num = jedis.lpush(queueKey.getBytes(), arrayOutputStream.toByteArray());
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
}

消費者

package queue.redisQueue;

import queue.fqueue.vo.EventVo;
import redis.clients.jedis.Jedis;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.List;

/**
 * @ProjectName: cutter-point
 * @Package: queue.redisQueue
 * @ClassName: RedisQueueConsume2
 * @Author: xiaof
 * @Description: ${description}
 * @Date: 2019/6/12 16:40
 * @Version: 1.0
 */
public class RedisQueueConsume2 implements Runnable {

    private Jedis jedis;
    private String queueKey;

    public RedisQueueConsume2(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }


    @Override
    public void run() {


        while(true) {
            List<byte[]> bytesList = null;
            try{
                //這種就是阻塞隊列模式
                bytesList = jedis.blpop(0, queueKey.getBytes());
            } catch (Exception e) {

            }

            //反序列化對象
            if(bytesList == null || bytesList.size() <= 0) {
                Thread.yield();
                continue;
            }

            //獲取第二個對象，就是我們的字節數組
            System.out.println(new String(bytesList.get(0)));
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytesList.get(1));
            try {
                ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
                EventVo eventVo = (EventVo) objectInputStream.readObject();

                eventVo.doOperater();

            } catch (IOException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
        }
    }
}