什么是消息队列？
消息（Message）是指在应用之间传送的数据，消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象。 消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回，有消息系统来确保信息的可靠传递，消息发布者只管把消息发布到MQ中而不管谁来取，消息使用者只管从MQ中取消息而不管谁发布的，这样发布者和使用者都不用知道对方的存在。
为什么使用消息队列？
从上面描述中可以看出消息队列是一种应用间的异步协作机制，那什么时候需要使用MQ呢？ 以常见的订单系统为例子，用户点击【下单】按钮后的业务逻辑包括：扣减库存、生成相应的单据、发红包、发短信通知‘在业务发展初期这些逻辑可能放在一起同步执行，随着业务订单量增长，需要提升系统服务的性能，这时候可以将一些不需要立即生效的操作拆分出来异步执行，，比如发红包、发短信通知等。这种场景就可以用MQ，在下单的主流程（比如扣减库存、生成相应的单据）完成之后发送一条消息到MQ让主流程快速完结，而由另外的单独线程拉取MQ的消息（或者由MQ推送消息），当发现MQ中有发红包或者发短信之类的消息，执行相应的业务逻辑。
使用消息队列的好处？
(1)提高系统的响应速度：
使用消息队列，生产者一方，把消息往消息队列里一扔，就可以立马返回响应用户，无需等待处理结果。
(2)保证消息的传递：
如果发送消息时接收者不可用，消息队列会保留消息，直到成功的传递它。
(3)解耦：
只要信息格式不变，即使接收者的接口，位置，或者配置改变，也不会给发送者带来任何改变。消息发送者无需知道消息接收者是谁，使得系统设计更清晰。
什么是延迟队列？
延迟队列，顾名思义它是一种带有延迟功能的消息队列。
延迟队列应用场景？
1.当订单一直处于未支付状态时，如何及时的关闭订单
2.如何定期检查处于退款状态的订单是否已经退款成功
3.在订单长时间没有收到下游系统的状态通知的时候，如何实现阶梯式的同步订单状态的策略
4.在系统通知上游系统支付成功终态时，上游系统返回通知失败，如何进行异步通知实行分频率发送:15s 3m 10m 30m 30m 1h 2h 6h 15h
解决方案：
1.最简单的方式，定时扫表。 例如对于订单支付失效要求比较高的，每2S扫表一次检查过期的订单进行主动关单操作。优点是简单，缺点是每分钟全局扫表，浪费资源，如果遇到表数据订单量即将过期的订单量很大，会造成关单延迟。
2.使用RabbitMq或者其他MQ改造实现延迟队列： 优点是，开源，现成的稳定的实现方案，缺点是：MQ是一个消息中间件，如果团队技术栈本来就有MQ，那还好，如果不是，那为了延迟队列而去部署一套MQ成本有点大
3.使用Redis实现延迟队列： 使用Redis的Hashtable、zset、list来实现延迟队列
这里我们可以先看一下 有赞的延迟队列设计
1.设计图
我们利用有赞的延迟队列设计进行改造
2.数据结构
ZING:DELAY_QUEUE:JOB_POOL： 是一个Hash_Table结构，里面存储了所有延迟队列的信息。KV结构：K=prefix+projectName,field=topic+jobId,V=CONENT;V由客户端传入的数据，消费的时候回传;
ZING:DELAY_QUEUE:BUCKET: 延迟队列的有序集合ZSET，存放K=ID和score为需要的执行时间戳，根据时间戳排序; ZING:DELAY_QUEUE:QUEUE LIST: 每个Topic一个LIST，list存放的都是当前需要被消费的JOB。
3.任务的生命周期
1.新增一个JOB，会在ZING:DELAY_QUEUE:JOB_POOL中插入一条数据，记录了业务方消费方。ZING:DELAY_QUEUE:BUCKET也会插入一条记录，记录执行的时间戳
2.搬运线程会去ZING:DELAY_QUEUE:BUCKET中查找哪些执行时间戳的RunTimeMillis比现在的时间小，将这些记录全部删除；同时会解析出每个任务的Topic是什么，然后将这些任务PUSH到TOPIC对应的列表ZING:DELAY_QUEUE:QUEUE中
3.每个TOPIC的LIST都会有一个监听线程去批量获取LIST中的待消费数据，获取到的数据全部扔给这个TOPIC的消费线程池
4.消费线程池执行会去ZING:DELAY_QUEUE:JOB_POOL查找数据结构，返回给回调结构，执行回调方法。
4.设计要点
4.1 基本概念
1.JOB：需要异步处理的任务，是延迟队列里的基本单元。
2.Topic：一组相同类型Job的集合（队列）。供消费者来订阅
4.2 消息结构
每个JOB必须包含以下几个属性：
jobId：Job的唯一标识。用来检索和删除指定的Job信息
topic：Job类型。可以理解成具体的业务名称
delay：Job需要延迟的时间。单位：秒。（服务端会将其转换为绝对时间）
body：Job的内容，供消费者做具体的业务处理，以json格式存储
retry：失败重试次数
url：通知URL
5.设计细节
如何快速消费ZING:DELAY_QUEUE:QUEUE
最简单的实现方式就是使用定时器进行秒级扫描，为了保证消息执行的时效性，可以设置每1S请求Redis一次，判断队列中是否有待消费的JOB。但是这样会存在一个问题，如果queue中一直没有可消费的JOB，那频繁的扫描就失去了意义，也浪费了资源，幸好LIST中有一个BLPOP阻塞原语，如果list中有数据就会立马返回,如果没有数据就会一直阻塞在那里，直到有数据返回,可以设置阻塞的超时时间,超时会返回NULL;具体的实现方式及策略会在代码中进行具体的实现介绍。
避免定时导致的消息重复搬运及消费
1.使用Redis的分布式锁来控制消息的搬运，从而避免消息被重复搬运导致的问题
2.使用分布式锁来保证定时器的执行频率。
6.核心代码实现
6.1 技术说明
技术栈： SpringBoot，Redisson，Redis，分布式锁，定时器
注意： 本项目没有实现设计方案中的多Queue消费，只开启了一个QUEUE，这个待以后优化
6.2 核心实体
6.2.1 Job新增对象
/**
* 消息结构
*
* @author 睁眼看世界
* @date 2020年1月15日
*/
@Data
public class Job implements Serializable {
private static final long serialVersionUID = 1L;
/**
* Job的唯一标识。用来检索和删除指定的Job信息
*/
@NotBlank
private String jobId;
/**
* Job类型。可以理解成具体的业务名称
*/
@NotBlank
private String topic;
/**
* Job需要延迟的时间。单位：秒。（服务端会将其转换为绝对时间）
*/
private Long delay;
/**
* Job的内容，供消费者做具体的业务处理，以json格式存储
*/
@NotBlank
private String body;
/**
* 失败重试次数
*/
private int retry = 0;
/**
* 通知URL
*/
@NotBlank
private String url;
}
6.2.2 Job删除对象
/**
* 消息结构
*
* @author 睁眼看世界
* @date 2020年1月15日
*/
@Data
public class JobDie implements Serializable {
private static final long serialVersionUID = 1L;
/**
* Job的唯一标识。用来检索和删除指定的Job信息
*/
@NotBlank
private String jobId;
/**
* Job类型。可以理解成具体的业务名称
*/
@NotBlank
private String topic;
}
6.2 搬运线程
/**
* 搬运线程
*
* @author 睁眼看世界
* @date 2020年1月17日
*/
@Slf4j
@Component
public class CarryJobScheduled {
@Autowired
private RedissonClient redissonClient;
/**
* 启动定时开启搬运JOB信息
*/
@Scheduled(cron = "*/1 * * * * *")
public void carryJobToQueue() {
System.out.println("carryJobToQueue --->");
RLock lock = redissonClient.getLock(RedisQueueKey.CARRY_THREAD_LOCK);
try {
boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
if (!lockFlag) {
throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
}
RScoredSortedSet<Object> bucketSet = redissonClient.getScoredSortedSet(RD_ZSET_BUCKET_PRE);
long now = System.currentTimeMillis();
Collection<Object> jobCollection = bucketSet.valueRange(0, false, now, true);
List<String> jobList = jobCollection.stream().map(String::valueOf).collect(Collectors.toList());
RList<String> readyQueue = redissonClient.getList(RD_LIST_TOPIC_PRE);
readyQueue.addAll(jobList);
bucketSet.removeAllAsync(jobList);
} catch (InterruptedException e) {
log.error("carryJobToQueue error", e);
} finally {
if (lock != null) {
lock.unlock();
}
}
}
}
6.4 消费线程
@Slf4j
@Component
public class ReadyQueueContext {
@Autowired
private RedissonClient redissonClient;
@Autowired
private ConsumerService consumerService;
/**
* TOPIC消费线程
*/
@PostConstruct
public void startTopicConsumer() {
TaskManager.doTask(this::runTopicThreads, "开启TOPIC消费线程");
}
/**
* 开启TOPIC消费线程
* 将所有可能出现的异常全部catch住，确保While(true)能够不中断
*/
@SuppressWarnings("InfiniteLoopStatement")
private void runTopicThreads() {
while (true) {
RLock lock = null;
try {
lock = redissonClient.getLock(CONSUMER_TOPIC_LOCK);
} catch (Exception e) {
log.error("runTopicThreads getLock error", e);
}
try {
if (lock == null) {
continue;
}
// 分布式锁时间比Blpop阻塞时间多1S，避免出现释放锁的时候，锁已经超时释放，unlock报错
boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
if (!lockFlag) {
continue;
}
// 1. 获取ReadyQueue中待消费的数据
RBlockingQueue<String> queue = redissonClient.getBlockingQueue(RD_LIST_TOPIC_PRE);
String topicId = queue.poll(60, TimeUnit.SECONDS);
if (StringUtils.isEmpty(topicId)) {
continue;
}
// 2. 获取job元信息内容
RMap<String, Job> jobPoolMap = redissonClient.getMap(JOB_POOL_KEY);
Job job = jobPoolMap.get(topicId);
// 3. 消费
FutureTask<Boolean> taskResult = TaskManager.doFutureTask(() -> consumerService.consumerMessage(job.getUrl(), job.getBody()), job.getTopic() + "-->消费JobId-->" + job.getJobId());
if (taskResult.get()) {
// 3.1 消费成功，删除JobPool和DelayBucket的job信息
jobPoolMap.remove(topicId);
} else {
int retrySum = job.getRetry() + 1;
// 3.2 消费失败，则根据策略重新加入Bucket
// 如果重试次数大于5，则将jobPool中的数据删除，持久化到DB
if (retrySum > RetryStrategyEnum.RETRY_FIVE.getRetry()) {
jobPoolMap.remove(topicId);
continue;
}
job.setRetry(retrySum);
long nextTime = job.getDelay() + RetryStrategyEnum.getDelayTime(job.getRetry()) * 1000;
log.info("next retryTime is [{}]", DateUtil.long2Str(nextTime));
RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
delayBucket.add(nextTime, topicId);
// 3.3 更新元信息失败次数
jobPoolMap.put(topicId, job);
}
} catch (Exception e) {
log.error("runTopicThreads error", e);
} finally {
if (lock != null) {
try {
lock.unlock();
} catch (Exception e) {
log.error("runTopicThreads unlock error", e);
}
}
}
}
}
}
6.5 添加及删除Job
/**
* 提供给外部服务的操作接口
*
* @author why
* @date 2020年1月15日
*/
@Slf4j
@Service
public class RedisDelayQueueServiceImpl implements RedisDelayQueueService {
@Autowired
private RedissonClient redissonClient;
/**
* 添加job元信息
*
* @param job 元信息
*/
@Override
public void addJob(Job job) {
RLock lock = redissonClient.getLock(ADD_JOB_LOCK + job.getJobId());
try {
boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
if (!lockFlag) {
throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
}
String topicId = RedisQueueKey.getTopicId(job.getTopic(), job.getJobId());
// 1. 将job添加到 JobPool中
RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
if (jobPool.get(topicId) != null) {
throw new BusinessException(ErrorMessageEnum.JOB_ALREADY_EXIST);
}
jobPool.put(topicId, job);
// 2. 将job添加到 DelayBucket中
RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
delayBucket.add(job.getDelay(), topicId);
} catch (InterruptedException e) {
log.error("addJob error", e);
} finally {
if (lock != null) {
lock.unlock();
}
}
}
/**
* 删除job信息
*
* @param job 元信息
*/
@Override
public void deleteJob(JobDie jobDie) {
RLock lock = redissonClient.getLock(DELETE_JOB_LOCK + jobDie.getJobId());
try {
boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
if (!lockFlag) {
throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
}
String topicId = RedisQueueKey.getTopicId(jobDie.getTopic(), jobDie.getJobId());
RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
jobPool.remove(topicId);
RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
delayBucket.remove(topicId);
} catch (InterruptedException e) {
log.error("addJob error", e);
} finally {
if (lock != null) {
lock.unlock();
}
}
}
}
京公网安备 11010502036488号