-
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;/*** Crea...展开全文
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;
}
}
weixin_39575170 2021-02-26 17:35:41 -
场景:1.获取一个实体对象;2.使用fastjson将实体对象转换为json字符串;3.将json字符串写入队列4.从队列中取出json字符串5.... 实体类CityBean.javapackage com.zbz.redis.bean;import java.io.Ser...展开全文
场景:
1.获取一个实体对象;
2.使用fastjson将实体对象转换为json字符串;
3.将json字符串写入队列
4.从队列中取出json字符串
5.将json字符串转换为对象
本例主要jar包:
jedis-2.9.0.jar;fastjson-1.2.41.jar
本例三个类:
1. 实体类CityBean.java
package com.zbz.redis.bean;
import java.io.Serializable;
import com.alibaba.fastjson.annotation.JSONField;
/**
* 城市信息
*
* @ClassName: CityBean
* @date: 2018-12-26 下午12:33:18
*/
public class CityBean implements Serializable {
private static final long serialVersionUID = 4443714978780088961L;
/** 城市名称 */
String CITY_NAME;
/** 电话区号 */
String AREA_CODE;
/** 邮政编码 */
String POSTAL_CODE;
/** 行政区划代码 */
String AREA_NUMBER;
/** 城市描述 */
String DESCRIBE;
@JSONField(name = "CITY_NAME")
public String getCITY_NAME() {
return CITY_NAME;
}
public void setCITY_NAME(String cITY_NAME) {
CITY_NAME = cITY_NAME;
}
@JSONField(name = "AREA_CODE")
public String getAREA_CODE() {
return AREA_CODE;
}
public void setAREA_CODE(String aREA_CODE) {
AREA_CODE = aREA_CODE;
}
@JSONField(name = "POSTAL_CODE")
public String getPOSTAL_CODE() {
return POSTAL_CODE;
}
public void setPOSTAL_CODE(String pOSTAL_CODE) {
POSTAL_CODE = pOSTAL_CODE;
}
@JSONField(name = "AREA_NUMBER")
public String getAREA_NUMBER() {
return AREA_NUMBER;
}
public void setAREA_NUMBER(String aREA_NUMBER) {
AREA_NUMBER = aREA_NUMBER;
}
@JSONField(name = "DESCRIBE")
public String getDESCRIBE() {
return DESCRIBE;
}
public void setDESCRIBE(String dESCRIBE) {
DESCRIBE = dESCRIBE;
}
@Override
public String toString() {
return "CityBean [CITY_NAME=" + CITY_NAME
+",AREA_CODE=" + AREA_CODE
+",POSTAL_CODE=" + POSTAL_CODE
+",AREA_NUMBER=" + AREA_NUMBER
+",DESCRIBE=" + DESCRIBE
+"]";
}
}
2. json转换类BeanUtils.java
package com.zbz.redis.utils;
import com.alibaba.fastjson.JSON;
/**
*
* @ClassName: BeanUtils
* @date: 2018-12-26 下午12:44:31
*/
public class BeanUtils {
/**实体对象转换为json字符串*/
public static String toJSONString(Object city){
String jsonStr = "";
if(city!=null){
jsonStr = JSON.toJSONString(city);
}
return jsonStr;
}
/**将json字符串转换为指定对象*/
public static T parseObject(String text, Classclazz) {
return JSON.parseObject(text, clazz);
}
}
3. redis操作类SimpleRedisClient.java
package com.zbz.redis.client;
import org.apache.commons.lang3.StringUtils;
import redis.clients.jedis.Jedis;
import com.zbz.redis.bean.CityBean;
import com.zbz.redis.utils.BeanUtils;
/**
*
* @ClassName: SimpleRedisClient
* @date: 2018-12-26 下午12:53:00
*/
public class SimpleRedisClient {
/**队列名称*/
public static final String cityQueue= "CITY_QUEUE";
/**redis服务器IP*/
public static final String redisIp= "127.0.0.1";
/**redis服务器端口*/
public static final int redisPort= 6379;
/**测试主函数
* @throws InterruptedException */
public static void main(String[] args) throws InterruptedException{
System.out.println("测试开始......");
writeRedisQueue();
System.out.println("主线程休眠10秒");
Thread.sleep(10000);
String jsonStr = ReadRedisQueue();
parseObject(jsonStr);
System.out.println("测试结束......");
}
public static void writeRedisQueue(){
System.out.println("测试写redis队列开始......");
String cityStr = getCity(getCityBean());
System.out.println("写入队列的数据: ");
System.out.println(cityStr);
/** 1.创建jedis实例,构造函数传入redis的IP地址和端口号 */
Jedis jedis = new Jedis(redisIp,redisPort);
/**2.字符串写入队列中*/
jedis.lpush(cityQueue,cityStr);
/** 3.关闭连接 */
jedis.close();
System.out.println("测试写redis队列结束......");
}
public static String ReadRedisQueue() {
System.out.println("测试读redis列开始......");
/** 1.创建jedis实例,构造函数传入redis的IP地址和端口号 */
Jedis jedis = new Jedis(redisIp,redisPort);
/** 2.在队列中读取数据*/
String result = jedis.lpop(cityQueue);
/** 3.关闭连接 */
jedis.close();
System.out.println("打印从队列获取的数据:");
System.out.println(result);
System.out.println("测试读redis列结束......");
return result;
}
/**json字符串转换为对象*/
public static void parseObject(String jsonStr){
if(StringUtils.isBlank(jsonStr)) return ;
CityBean cityBean = BeanUtils.parseObject(jsonStr, CityBean.class);
System.out.println("输出从字符串转换后的对象:");
System.out.println(cityBean.toString());
}
/**获取城市json字符串*/
public static String getCity(CityBean city){
return BeanUtils.toJSONString(city);
}
/**获取一个城市对象*/
public static CityBean getCityBean(){
CityBean cityBean = new CityBean();
cityBean.setCITY_NAME("厦门");
cityBean.setAREA_CODE("0592");
cityBean.setPOSTAL_CODE("361000");
cityBean.setAREA_NUMBER("350200");
cityBean.setDESCRIBE("适合居住");
return cityBean;
}
}
测试:
1.在redis-server安装目录双击启动redis-server.exe服务端
2.在redis-cli安装目录双击启动redis-cli.exe客户端
3.使用 Llen CITY_QUEUE 命令,在redis-cli客户端查看队列长度变化
提示:
redis队列,写入数据时,队列存在则往队列写数据,不存在就创建再往队列写;
取数据时,取出一个数据,队列数据就少一个,最后一个取出后,队列会消失,
当队列没数据时去取数据,会返回null.
截图:
1.redis-server截图
2.redis-cli.exe截图
3.控制台输出日志
以上,TKS.
weixin_33735910 2021-04-22 06:59:06 -
packagecom.yys.demo.config;importorg.springframework.beans.factory.annotation....importorg.springframework.data.redis.core.RedisTemplate;importorg.springframework.stereotype.Component;importorg...展开全文
packagecom.yys.demo.config;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.redis.core.RedisTemplate;importorg.springframework.stereotype.Component;importorg.springframework.util.CollectionUtils;importjava.util.List;importjava.util.Map;importjava.util.Set;importjava.util.concurrent.TimeUnit;/*** 描述：redis工具类(本配置类只有redis消息队列相关命令方法)
*@authoryys
* @date 2019.03.15*/@Componentpublic classRedisClient {
@Autowiredprivate RedisTemplateredisTemplate;/**---------------------------------- redis消息队列 ----------------------------------*/
/*** 存值
*@paramkey 键
*@paramvalue 值
*@return
*/
public booleanlpush(String key, Object value) {try{
redisTemplate.opsForList().leftPush(key, value);return true;
}catch(Exception e) {
e.printStackTrace();return false;
}
}/*** 取值 -
*@paramkey 键
*@return
*/
publicObject rpop(String key) {try{returnredisTemplate.opsForList().rightPop(key);
}catch(Exception e) {
e.printStackTrace();return null;
}
}/*** 取值 - - 推荐使用
*@paramkey 键
*@paramtimeout 超时时间
*@paramtimeUnit 给定单元粒度的时间段
* TimeUnit.DAYS //天
* TimeUnit.HOURS //小时
* TimeUnit.MINUTES //分钟
* TimeUnit.SECONDS //秒
* TimeUnit.MILLISECONDS //毫秒
*@return
*/
public Object brpop(String key, longtimeout, TimeUnit timeUnit) {try{returnredisTemplate.opsForList().rightPop(key, timeout, TimeUnit.SECONDS);
}catch(Exception e) {
e.printStackTrace();return null;
}
}/*** 查看值
*@paramkey 键
*@paramstart 开始
*@paramend 结束 0 到 -1代表所有值
*@return
*/
public List lrange(String key, long start, longend) {try{returnredisTemplate.opsForList().range(key, start, end);
}catch(Exception e) {
e.printStackTrace();return null;
}
}/**---------------------------------- redis消息队列 ----------------------------------*/}
weixin_29424569 2021-03-06 00:55:22 -
Java - Redis 实现延时消息队列什么是延时任务延时任务的特点实现思路：代码实现使用事例 什么是延时任务 延时任务，顾名思义，就是延迟一段时间后才执行的任务。举个例子，假设我们有个发布资讯的功能，运营需要在...展开全文
什么是延时任务
延时任务，顾名思义，就是延迟一段时间后才执行的任务。举个例子，假设我们有个发布资讯的功能，运营需要在每天早上7点准时发布资讯，但是早上7点大家都还没上班，这个时候就可以使用延时任务来实现资讯的延时发布了。只要在前一天下班前指定第二天要发送资讯的时间，到了第二天指定的时间点资讯就能准时发出去了。如果大家有运营过公众号，就会知道公众号后台也有文章定时发送的功能。总而言之，延时任务的使用还是很广泛的。
延时任务的特点
- 时间有序性
- 时间具体性
- 任务中携带详细的信息 ，通常包括 任务ID， 任务的类型 ，时间点。
实现思路：
将整个Redis当做消息池，以kv形式存储消息，key为id，value为具体的消息body
使用ZSET做优先队列，按照score维持优先级（用当前时间+需要延时的时间作为score）
轮询ZSET，拿出score比当前时间戳大的数据（已过期的）
根据id拿到消息池的具体消息进行消费
消费成功，删除改队列和消息
消费失败，让该消息重新回到队列代码实现
1.消息模型import lombok.Data; import lombok.experimental.Accessors; import javax.validation.constraints.NotNull; import java.io.Serializable; /** * Redis 消息队列中的消息体 * @author shikanatsu */ @Data @Accessors(chain = true) public class RedisMessage implements Serializable { /** 消息队列组 **/ private String group; /** * 消息id */ private String id; /** * 消息延迟/ 秒 */ @NotNull(message = "消息延时时间不能为空") private long delay; /** * 消息存活时间 单位：秒 */ @NotNull(message = "消息存活时间不能为空") private int ttl; /** * 消息体，对应业务内容 */ private Object body; /** * 创建时间，如果只有优先级没有延迟，可以设置创建时间为0 * 用来消除时间的影响 */ private long createTime; }RedisMq 消息队列实现类package com.shixun.base.redisMq; import com.shixun.base.jedis.service.RedisService; import org.springframework.stereotype.Component; import javax.annotation.Resource; /** * Redis消息队列 * * @author shikanatsu */ @Component public class RedisMq { /** * 消息池前缀，以此前缀加上传递的消息id作为key，以消息{@link MSG_POOL} * 的消息体body作为值存储 */ public static final String MSG_POOL = "Message:Pool:"; /** * zset队列 名称 queue */ public static final String QUEUE_NAME = "Message:Queue:"; // private static final int SEMIH = 30 * 60; @Resource private RedisService redisService; /** * 存入消息池 * * @param message * @return */ public boolean addMsgPool(RedisMessage message) { if (null != message) { redisService.set(MSG_POOL + message.getGroup() + message.getId(), message, message.getTtl()); return true; } return false; } /** * 从消息池中删除消息 * * @param id * @return */ public void deMsgPool(String group, String id) { redisService.remove(MSG_POOL + group + id); } /** * 向队列中添加消息 * * @param key * @param score 优先级 * @param val * @return 返回消息id */ public void enMessage(String key, long score, String val) { redisService.zsset(key, val, score); } /** * 从队列删除消息 * * @param id * @return */ public boolean deMessage(String key, String id) { return redisService.zdel(key, id); } }3.消息生产者import cn.hutool.core.convert.Convert; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.IdUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import org.springframework.validation.annotation.Validated; import javax.annotation.Resource; import java.text.SimpleDateFormat; import java.util.Date; import java.util.concurrent.TimeUnit; /** * 消息生产者 * * @author shikanatsu */ @Component public class MessageProvider { static Logger logger = LoggerFactory.getLogger(MessageProvider.class); @Resource private RedisMq redisMq; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); public boolean sendMessage(@Validated RedisMessage message) { Assert.notNull(message); //The priority is if there is no creation time // message.setCreateTime(System.currentTimeMillis()); message.setId(IdUtil.fastUUID()); Long delayTime = message.getCreateTime() + Convert.convertTime(message.getDelay(), TimeUnit.SECONDS, TimeUnit.MILLISECONDS); try { redisMq.addMsgPool(message); redisMq.enMessage(RedisMq.QUEUE_NAME+message.getGroup(), delayTime, message.getId()); logger.info("RedisMq发送消费信息{}，当前时间:{},消费时间预计{}",message.toString(),new Date(),sdf.format(delayTime)); }catch (Exception e){ e.printStackTrace(); logger.error("RedisMq 消息发送失败，当前时间:{}",new Date()); return false; } return true; } }4.消息消费者/** * Redis消息消费者 * @author shikanatsu */ @Component public class RedisMqConsumer { private static final Logger log = LoggerFactory.getLogger(RedisMqConsumer.class); @Resource private RedisMq redisMq; @Resource private RedisService redisService; @Resource private MessageProvider provider; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); //@Scheduled(cron = "*/1 * * * * ? ") /** Instead of a thread loop, you can use Cron expressions to perform periodic tasks */ public void baseMonitor(RedisMqExecute mqExecute){ String queueName = RedisMq.QUEUE_NAME+mqExecute.getQueueName(); //The query is currently expired Set<Object> set = redisService.rangeByScore(queueName, 0, System.currentTimeMillis()); if (null != set) { long current = System.currentTimeMillis(); for (Object id : set) { long score = redisService.getScore(queueName, id.toString()).longValue(); //Once again the guarantee has expired ， And then perform the consumption if (current >= score) { String str = ""; RedisMessage message = null; String msgPool = RedisMq.MSG_POOL+mqExecute.getQueueName(); try { message = (RedisMessage)redisService.get(msgPool + id.toString()); log.debug("RedisMq:{},get RedisMessage success now Time:{}",str,sdf.format(System.currentTimeMillis())); if(null==message){ return; } //Do something ； You can add a judgment here and if it fails you can add it to the queue again mqExecute.execute(message); } catch (Exception e) { e.printStackTrace(); //If an exception occurs, it is put back into the queue // todo： If repeated, this can lead to repeated cycles log.error("RedisMq: RedisMqMessage exception ,It message rollback , If repeated, this can lead to repeated cycles{}",new Date()); provider.sendMessage(message); } finally { redisMq.deMessage(queueName, id.toString()); redisMq.deMsgPool(message.getGroup(),id.toString()); } } } } } }5. 消息执接口/** * @author shikanatsu */ public interface RedisMqExecute { /** * 获取队列名称 * @return */ public String getQueueName(); /** * 统一的通过执行期执行 * @param message * @return */ public boolean execute(RedisMessage message); /** * Perform thread polling */ public void threadPolling(); }6. 任务类型的实现类：可以根据自己的情况去实现对应的队列需求/** * 订单执行 * * @author shikanatsu */ @Service public class OrderMqExecuteImpl implements RedisMqExecute { private static Logger logger = LoggerFactory.getLogger(OrderMqExecuteImpl.class); public final static String name = "orderPoll:"; @Resource private RedisMqConsumer redisMqConsumer; private RedisMqExecute mqExecute = this; @Resource private OrderService orderService; @Override public String getQueueName() { return name; } @Override /** * For the time being, only all orders will be processed. You can change to make orders */ public boolean execute(RedisMessage message) { logger.info("Do orderMqPoll ; Time:{}",new Date()); //Do return true; } @Override /** 通过线程去执行轮询的过程，时间上可以自由控制 **/ public void threadPolling() { ThreadUtil.execute(() -> { while (true) { redisMqConsumer.baseMonitor(mqExecute); ThreadUtil.sleep(5, TimeUnit.MICROSECONDS); } }); } }使用事例
1. 实现RedisMqExecute 接口 创建对应的轮询或者采取定时器的方式执行 和实现具体的任务。 2. 通过MessageProvider 实现相对应的消息服务和绑定队列组，通过队列组的方式执行。 3. 提示： 采取线程的方式需要在项目启动过程中执行，采取定时器或者调度的方式可以更加动态的调整。
qq_26093077 2021-01-26 15:04:46 -
weixin_29863401 2021-03-04 02:10:10
-
weixin_39926193 2021-02-12 14:48:51
-
weixin_42146230 2021-02-28 07:58:35
-
weixin_35260789 2021-02-12 14:44:34
-
qq_38234785 2021-07-01 11:16:00
-
weixin_39810856 2021-02-12 14:46:42
-
weixin_44742132 2021-01-12 11:48:51
-
gubeichengxuyuan 2021-11-05 09:28:32
-
weixin_42512936 2021-02-28 07:53:07
-
weixin_35480412 2021-02-13 00:18:53
-
weixin_39855568 2020-12-21 14:12:32
-
weixin_42347422 2021-03-21 08:48:28
-
weixin_42502345 2021-02-12 14:44:35
-
weixin_39740419 2021-03-17 14:56:40
-
weixin_36349918 2021-02-28 17:36:45
-
weixin_42509396 2021-02-28 17:36:44
-
weixin_29648175 2021-03-21 11:11:06
-
weixin_33309848 2021-03-15 01:59:03
-
weixin_31880383 2021-03-14 21:21:06
-
jack1liu 2021-02-06 14:54:32
-
weixin_33510225 2021-03-16 16:22:14
-
lxw1844912514 2021-01-14 19:11:43
java
订阅
redis
订阅