Java利用Redis实现消息队列

素颜马尾好姑娘i 2021-12-20 14:43 318阅读 0赞

原文地址：[Java利用Redis实现消息队列][Java_Redis]

以下为转载------------------------------------------------------------

### 应用场景 ###

*  为什么要用redis?  
    二进制存储、java序列化传输、IO连接数高、连接频繁

### 一、序列化 ###

这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口;   
其代码如下:

package Utils;
    import java.io.*;
    /**
     * Created by Kinglf on 2016/10/17.
     */
    public class ObjectUtil {
        /**
         * 对象转byte[]
         * @param obj
         * @return
         * @throws IOException
         */
        public static byte[] object2Bytes(Object obj) throws IOException{
            ByteArrayOutputStream bo=new ByteArrayOutputStream();
            ObjectOutputStream oo=new ObjectOutputStream(bo);
            oo.writeObject(obj);
            byte[] bytes=bo.toByteArray();
            bo.close();
            oo.close();
            return bytes;
        }
        /**
         * byte[]转对象
         * @param bytes
         * @return
         * @throws Exception
         */
        public static Object bytes2Object(byte[] bytes) throws Exception{
            ByteArrayInputStream in=new ByteArrayInputStream(bytes);
            ObjectInputStream sIn=new ObjectInputStream(in);
            return sIn.readObject();
        }
    }

### 二、消息类(实现Serializable接口) ###

package Model;
    
    import java.io.Serializable;
    
    /**
     * Created by Kinglf on 2016/10/17.
     */
    public class Message implements Serializable {
    
        private static final long serialVersionUID = -389326121047047723L;
        private int id;
        private String content;
        public Message(int id, String content) {
            this.id = id;
            this.content = content;
        }
        public int getId() {
            return id;
        }
        public void setId(int id) {
            this.id = id;
        }
        public String getContent() {
            return content;
        }
        public void setContent(String content) {
            this.content = content;
        }
    }

### 三、Redis的操作 ###

利用redis做队列,我们采用的是redis中list的push和pop操作;   
结合队列的特点:   
`只允许在一端插入新元素只能在队列的尾部FIFO:先进先出原则` Redis中lpush`头入`(rpop`尾出`)或rpush`尾入`(lpop`头出`)可以满足要求,而Redis中list药push或　pop的对象仅需要转换成byte\[\]即可   
`　　java采用Jedis进行Redis的存储和Redis的连接池设置`   
上代码:

package Utils;
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.JedisPoolConfig;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    /**
     * Created by Kinglf on 2016/10/17.
     */
    public class JedisUtil {
        private static String JEDIS_IP;
        private static int JEDIS_PORT;
        private static String JEDIS_PASSWORD;
        private static JedisPool jedisPool;
        static {
            //Configuration自行写的配置文件解析类,继承自Properties
            Configuration conf=Configuration.getInstance();
            JEDIS_IP=conf.getString("jedis.ip","127.0.0.1");
            JEDIS_PORT=conf.getInt("jedis.port",6379);
            JEDIS_PASSWORD=conf.getString("jedis.password",null);
            JedisPoolConfig config=new JedisPoolConfig();
            config.setMaxActive(5000);
            config.setMaxIdle(256);
            config.setMaxWait(5000L);
            config.setTestOnBorrow(true);
            config.setTestOnReturn(true);
            config.setTestWhileIdle(true);
            config.setMinEvictableIdleTimeMillis(60000L);
            config.setTimeBetweenEvictionRunsMillis(3000L);
            config.setNumTestsPerEvictionRun(-1);
            jedisPool=new JedisPool(config,JEDIS_IP,JEDIS_PORT,60000);
        }
        /**
         * 获取数据
         * @param key
         * @return
         */
        public static String get(String key){
            String value=null;
            Jedis jedis=null;
            try{
                jedis=jedisPool.getResource();
                value=jedis.get(key);
            }catch (Exception e){
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            }finally {
                close(jedis);
            }
            return value;
        }
    
        private static void close(Jedis jedis) {
            try{
                jedisPool.returnResource(jedis);
            }catch (Exception e){
                if(jedis.isConnected()){
                    jedis.quit();
                    jedis.disconnect();
                }
            }
        }
        public static byte[] get(byte[] key){
            byte[] value = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                value = jedis.get(key);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
    
            return value;
        }
    
        public static void set(byte[] key, byte[] value) {
    
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.set(key, value);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        public static void set(byte[] key, byte[] value, int time) {
    
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.set(key, value);
                jedis.expire(key, time);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        public static void hset(byte[] key, byte[] field, byte[] value) {
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.hset(key, field, value);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        public static void hset(String key, String field, String value) {
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.hset(key, field, value);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        /**
         * 获取数据
         *
         * @param key
         * @return
         */
        public static String hget(String key, String field) {
    
            String value = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                value = jedis.hget(key, field);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
    
            return value;
        }
        /**
         * 获取数据
         *
         * @param key
         * @return
         */
        public static byte[] hget(byte[] key, byte[] field) {
    
            byte[] value = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                value = jedis.hget(key, field);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
    
            return value;
        }
        public static void hdel(byte[] key, byte[] field) {
    
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.hdel(key, field);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
        /**
         * 存储REDIS队列 顺序存储
         * @param  key reids键名
         * @param  value 键值
         */
        public static void lpush(byte[] key, byte[] value) {
    
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.lpush(key, value);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        /**
         * 存储REDIS队列 反向存储
         * @param  key reids键名
         * @param  value 键值
         */
        public static void rpush(byte[] key, byte[] value) {
    
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                jedis.rpush(key, value);
    
            } catch (Exception e) {
    
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
    
                //返还到连接池
                close(jedis);
    
            }
        }
    
        /**
         * 将列表 source 中的最后一个元素(尾元素)弹出，并返回给客户端
         * @param  key reids键名
         * @param  destination 键值
         */
        public static void rpoplpush(byte[] key, byte[] destination) {
    
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                jedis.rpoplpush(key, destination);
    
            } catch (Exception e) {
    
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
    
                //返还到连接池
                close(jedis);
    
            }
        }
    
        /**
         * 获取队列数据
         * @param  key 键名
         * @return
         */
        public static List lpopList(byte[] key) {
    
            List list = null;
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                list = jedis.lrange(key, 0, -1);
    
            } catch (Exception e) {
    
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
    
                //返还到连接池
                close(jedis);
    
            }
            return list;
        }
        /**
         * 获取队列数据
         * @param  key 键名
         * @return
         */
        public static byte[] rpop(byte[] key) {
    
            byte[] bytes = null;
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                bytes = jedis.rpop(key);
    
            } catch (Exception e) {
    
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
    
                //返还到连接池
                close(jedis);
    
            }
            return bytes;
        }
        public static void hmset(Object key, Map hash) {
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.hmset(key.toString(), hash);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
    
            }
        }
        public static void hmset(Object key, Map hash, int time) {
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                jedis.hmset(key.toString(), hash);
                jedis.expire(key.toString(), time);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
    
            }
        }
        public static List hmget(Object key, String... fields) {
            List result = null;
            Jedis jedis = null;
            try {
    
                jedis = jedisPool.getResource();
                result = jedis.hmget(key.toString(), fields);
    
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
    
            }
            return result;
        }
    
        public static Set hkeys(String key) {
            Set result = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                result = jedis.hkeys(key);
    
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
    
            }
            return result;
        }
        public static List lrange(byte[] key, int from, int to) {
            List result = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                result = jedis.lrange(key, from, to);
    
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
    
            }
            return result;
        }
        public static Map hgetAll(byte[] key) {
            Map result = null;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                result = jedis.hgetAll(key);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
    
            } finally {
                //返还到连接池
                close(jedis);
            }
            return result;
        }
    
        public static void del(byte[] key) {
    
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.del(key);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
        }
    
        public static long llen(byte[] key) {
    
            long len = 0;
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                jedis.llen(key);
            } catch (Exception e) {
                //释放redis对象
                jedisPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                //返还到连接池
                close(jedis);
            }
            return len;
        }
    }

### 四、Configuration主要用于读取Redis的配置信息 ###

package Utils;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.Properties;
    
    /**
     * Created by Kinglf on 2016/10/17.
     */
    public class Configuration extends Properties {
    
        private static final long serialVersionUID = -2296275030489943706L;
        private static Configuration instance = null;
    
        public static synchronized Configuration getInstance() {
            if (instance == null) {
                instance = new Configuration();
            }
            return instance;
        }
    
    
        public String getProperty(String key, String defaultValue) {
            String val = getProperty(key);
            return (val == null || val.isEmpty()) ? defaultValue : val;
        }
    
        public String getString(String name, String defaultValue) {
            return this.getProperty(name, defaultValue);
        }
    
        public int getInt(String name, int defaultValue) {
            String val = this.getProperty(name);
            return (val == null || val.isEmpty()) ? defaultValue : Integer.parseInt(val);
        }
    
        public long getLong(String name, long defaultValue) {
            String val = this.getProperty(name);
            return (val == null || val.isEmpty()) ? defaultValue : Integer.parseInt(val);
        }
    
        public float getFloat(String name, float defaultValue) {
            String val = this.getProperty(name);
            return (val == null || val.isEmpty()) ? defaultValue : Float.parseFloat(val);
        }
    
        public double getDouble(String name, double defaultValue) {
            String val = this.getProperty(name);
            return (val == null || val.isEmpty()) ? defaultValue : Double.parseDouble(val);
        }
    
        public byte getByte(String name, byte defaultValue) {
            String val = this.getProperty(name);
            return (val == null || val.isEmpty()) ? defaultValue : Byte.parseByte(val);
        }
    
        public Configuration() {
            InputStream in = ClassLoader.getSystemClassLoader().getResourceAsStream("config.xml");
            try {
                this.loadFromXML(in);
                in.close();
            } catch (IOException ioe) {
    
            }
        }
    }

### 五、测试 ###

import Model.Message;
    import Utils.JedisUtil;
    import Utils.ObjectUtil;
    import redis.clients.jedis.Jedis;
    
    import java.io.IOException;
    
    /**
     * Created by Kinglf on 2016/10/17.
     */
    public class TestRedisQueue {
        public static byte[] redisKey = "key".getBytes();
        static {
            try {
                init();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        private static void init() throws IOException {
            for (int i = 0; i < 1000000; i++) {
                Message message = new Message(i, "这是第" + i + "个内容");
                JedisUtil.lpush(redisKey, ObjectUtil.object2Bytes(message));
            }
    
        }
    
        public static void main(String[] args) {
            try {
                pop();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private static void pop() throws Exception {
            byte[] bytes = JedisUtil.rpop(redisKey);
            Message msg = (Message) ObjectUtil.bytes2Object(bytes);
            if (msg != null) {
                System.out.println(msg.getId() + "----" + msg.getContent());
            }
        }
    }

每执行一次pop()方法,结果如下:
    <br>1----这是第1个内容
    <br>2----这是第2个内容
    <br>3----这是第3个内容
    <br>4----这是第4个内容

### 总结 ###

`至此,整个Redis消息队列的生产者和消费者代码已经完成`

1.  Message `需要传送的实体类(需实现Serializable接口)`
2.  Configuration `Redis的配置读取类,继承自Properties`
3.  ObjectUtil `将对象和byte数组双向转换的工具类`
4.  Jedis `通过消息队列的先进先出(FIFO)的特点结合Redis的list中的push和pop操作进行封装的工具类`

[Java_Redis]: https://www.cnblogs.com/kinglf/p/5972300.html