Analyse spécifique de la façon dont Java utilise Redis pour implémenter les files d'attente de messages

Cet article présente principalement l'exemple de code Java utilisant Redis pour implémenter la file d'attente de messages. Je pense que c'est plutôt bien, je vais donc le partager avec vous maintenant et le donner comme référence. Suivons l'éditeur et jetons un coup d'œil.

Cet article présente l'exemple de code de Java utilisant Redis pour implémenter la file d'attente de messages. J'aimerais le partager avec vous. Les détails sont les suivants :

Scénarios d'application

Pourquoi utiliser Redis ?

Stockage binaire, transmission sérialisée Java, nombre élevé de connexions IO, connexions fréquentes

1. Sérialisation

Un outil de sérialisation Java est écrit ici, qui convertit principalement les objets en tableaux d'octets et les désérialise en objets Java en fonction des tableaux d'octets ; utilisant principalement ByteArrayOutputStream et ByteArrayInputStream ; Remarque : chaque objet qui doit être sérialisé doit implémenter l'interface Serialisable

Le code est le suivant :

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();
 }
}

2. Classe de message (implémentant l'interface sérialisable)

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;
 }
}

3. Opération Redis

En utilisant redis comme file d'attente, nous utilisons les opérations push et pop de la liste dans redis

Combinées ; avec les caractéristiques de la file d'attente :

Dans Redis, lpush head-in (rpop tail-out) ou rpush tail-in (lpop head-out) peut répondre aux exigences , tandis que l'objet de list medicine push ou pop dans Redis n'a besoin que d'être converti en octet [] peut être 只允许在一端插入新元素只能在队列的尾部FIFO:先进先出原则

java采用Jedis进行Redis的存储和Redis的连接池设置

et ajouter le code :

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;
 }
}

Quatrièmement,La configuration est principalement utilisée pour lire les informations de 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) {

  }
 }
}

5. Tests

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个内容

Résumé

Ça y est, le code producteur et consommateur de l'ensemble de la file d'attente de messages Redis a été complété

Classe d'entité de message. qui doit être transmis (doit implémenter l'interface Serialisable)

2.Configuration Redis Classe de lecture de configuration, héritée de Properties

3.ObjectUtil Classe utilitaire pour la conversion bidirectionnelle d'objets et de tableaux d'octets

4.Les fonctionnalités Jedis premier entré, premier sorti via la file d'attente de messages (FIFO) combinées aux opérations push et pop dans la liste Redis sont des classes d'outils encapsulées

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!