Java利用Redis实现消息队列

矫情吗；* 2022-05-17 22:41 228阅读 0赞

### 一、序列化 ###

这里编写了一个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();